diff --git a/README.md b/README.md index c7b2fc6..6327fd7 100644 --- a/README.md +++ b/README.md @@ -236,9 +236,8 @@ A change that does not reach the schedule (e.g. a tag's `description`) does not #### Deploying the tag database write tags 1. Run the migration that adds `data_type` to the `connector_tags` validator (`sientia-dataops-database-migrations`). -2. Paste `scripts/sientiapde_1974_backfill_write_tags.js` into the MongoDB Compass shell (or mongosh), with `ACTION = 'backfill'`: first with `DRY_RUN = true` (the default) to see the plan, then with `DRY_RUN = false`. It creates the `connector_tags` of the legacy write tags, rewrites `write_tags` as `[{tag_id}]` (keeping the original in `legacy_write_tags`) and bumps the pipelines' `updated_at`. `ACTION = 'revert'` undoes it. +2. Make sure every `predictions_batch` / `xgboost_predictions_batch` declares its write tags as `[{tag_id}]`, with each tag in `connector_tags` (category `Write` or `Prediction Confidence`, and its `data_type`). Entries still in the legacy form (`server_id` / `type` / `addr`) are left out and notified. Convert them together with the deploy: the previous version cannot read `[{tag_id}]`. 3. Deploy this version. Its first run redeploys every schedule once, to record their `config_hash` (see above). -4. Once the write tags are confirmed, run the script with `ACTION = 'cleanup'` to remove `legacy_write_tags`. After that, `revert` can no longer restore the pipelines. #### Architecture diff --git a/scripts/sientiapde_1974_backfill_write_tags.js b/scripts/sientiapde_1974_backfill_write_tags.js deleted file mode 100644 index 6f40bee..0000000 --- a/scripts/sientiapde_1974_backfill_write_tags.js +++ /dev/null @@ -1,284 +0,0 @@ -/** - * SIENTIAPDE-1974 runbook: move the predictions_batch write tags to the tag database. - * - * The Orchestrator reads write_tags as [{tag_id}] only: server, address, category and - * data_type come from connector_tags + connectors. This script converts the legacy write - * tags ({server_id, type, addr, data_type}): - * - server_id (opc-servers id) -> connector: the connector whose legacy_opc_server_id is - * the server_id, else the one whose connector_id is the slug of the server name (D-D), - * else the one whose connector_id is the server_id itself; - * - each tag -> the connector_tags document with the same connector, address and category - * (type prediction -> Write, confidence -> Prediction Confidence), created when missing - * with the name :[:n] (D-A) and the tag's data_type; - * - write_tags -> [{tag_id}], the original array kept in legacy_write_tags, and - * updated_at bumped (D-C) so the Orchestrator redeploys the schedule. - * Entries that already carry a tag_id are reduced to {tag_id}; their data_type, when the - * pipeline sets one, is copied to the connector_tags document that has none (a different - * one aborts). - * - * Every check runs before the first write: any conflict aborts with nothing written. - * Idempotent: a converted pipeline has no legacy entry left and is skipped. - * - * Usage: paste the whole file into the MongoDB Compass shell (_MONGOSH) or mongosh, after - * setting ACTION and DRY_RUN below. It runs against DB_NAME whatever database the shell - * is on, and can be pasted again in the same shell. - * 1. ACTION 'backfill', DRY_RUN true: prints the plan, writes nothing. - * 2. ACTION 'backfill', DRY_RUN false: converts the pipelines. - * 3. Once the new write tags are confirmed in production, ACTION 'cleanup', DRY_RUN - * false: removes legacy_write_tags from the pipelines. After it, 'revert' can no - * longer restore them. - * ACTION 'revert', DRY_RUN false: restores legacy_write_tags into write_tags and deletes - * the connector_tags this script created that no other pipeline references. It only - * restores pipelines that still have legacy_write_tags (not cleaned up). - */ - -(() => { -// ---- settings ---------------------------------------------------------------------- -const ACTION = 'backfill'; // 'backfill' | 'revert' | 'cleanup' -const DRY_RUN = true; // false writes -const DB_NAME = 'sientia'; -// ------------------------------------------------------------------------------------- - -const db = globalThis.db.getSiblingDB(DB_NAME); -const SOURCE = 'backfill-SIENTIAPDE-1974-write-tags'; -const WRITE_TYPES = ['predictions_batch', 'xgboost_predictions_batch']; -const WRITE_CATEGORY = { prediction: 'Write', confidence: 'Prediction Confidence' }; -const DATA_TYPES = ['float', 'double', 'real', 'int', 'integer', 'bool', 'boolean', 'str', 'string']; - -const hasValue = (value) => value !== undefined && value !== null && value !== ''; - -// D-D: connector_id is the slug of the server name (same rule as the tag database backfill) -const slugify = (name) => - String(name) - .normalize('NFKD') - .replace(/[^\x00-\x7f]/g, '') - .toLowerCase() - .replace(/[^a-z0-9]+/g, '-') - .replace(/^-+|-+$/g, ''); - -// D-A: write tags have no name: :, plus : when the type repeats -const writeTagName = (schedule, tags, index) => { - const same = tags.flatMap((tag, i) => (tag.type === tags[index].type ? [i] : [])); - const base = `${schedule}:${tags[index].type}`; - return same.length === 1 ? base : `${base}:${same.indexOf(index) + 1}`; -}; - -const connectorResolver = () => { - const connectors = db.connectors.find({ type: 'opc_ua' }).toArray(); - const byLegacyId = new Map( - connectors.filter((c) => hasValue(c.legacy_opc_server_id)).map((c) => [String(c.legacy_opc_server_id), c]) - ); - const byId = new Map(connectors.map((c) => [String(c.connector_id), c])); - const servers = new Map(db.getCollection('opc-servers').find({}).toArray().map((s) => [String(s.id), s])); - return (serverId) => { - const sid = String(serverId); - if (byLegacyId.has(sid)) return byLegacyId.get(sid); - const server = servers.get(sid); - if (server && hasValue(server.server_name) && byId.has(slugify(server.server_name))) { - return byId.get(slugify(server.server_name)); - } - return byId.get(sid); - }; -}; - -const buildPlan = () => { - const errors = []; - const connectorOf = connectorResolver(); - const tags = db.connector_tags.find({}).toArray(); - const byAddress = new Map(tags.map((t) => [`${t.connector_id}|${t.address}|${t.category}`, t])); - const byName = new Map(tags.map((t) => [`${t.connector_id}|${t.tag_name}`, t])); - const byTagId = new Map(tags.map((t) => [String(t.tag_id), t])); - const newTags = []; - const dataTypes = new Map(); // tag_id -> data_type to set on an existing connector_tags - const updates = []; - - const pipelines = db.pipelines.find({ workflow_type: { $in: WRITE_TYPES } }).toArray(); - for (const pipeline of pipelines) { - const legacy = pipeline.write_tags; - if (!Array.isArray(legacy) || legacy.every((t) => hasValue(t?.tag_id) && Object.keys(t).length === 1)) { - continue; // nothing to convert - } - const refs = []; - legacy.forEach((tag, index) => { - const where = `${pipeline.schedule_name} write_tags[${index}]`; - if (hasValue(tag?.tag_id)) { - const existing = byTagId.get(String(tag.tag_id)); - if (!existing) return errors.push(`${where}: tag_id ${tag.tag_id} not in connector_tags`); - if (hasValue(tag.data_type)) { - const current = existing.data_type ?? dataTypes.get(String(tag.tag_id)); - if (!DATA_TYPES.includes(tag.data_type)) { - return errors.push(`${where}: data_type ${tag.data_type} not accepted`); - } - if (hasValue(current) && current !== tag.data_type) { - return errors.push(`${where}: tag_id ${tag.tag_id} has data_type ${current}, the pipeline ${tag.data_type}`); - } - if (!hasValue(existing.data_type)) dataTypes.set(String(tag.tag_id), tag.data_type); - } - refs.push({ tag_id: existing.tag_id }); - return; - } - const connector = connectorOf(tag?.server_id); - const category = WRITE_CATEGORY[tag?.type]; - if (!connector) return errors.push(`${where}: no opc_ua connector for server_id ${tag?.server_id}`); - if (!category) return errors.push(`${where}: type ${tag?.type} is neither prediction nor confidence`); - if (!hasValue(tag.addr)) return errors.push(`${where}: no addr`); - const dataType = tag.data_type ?? 'float'; - if (!DATA_TYPES.includes(dataType)) return errors.push(`${where}: data_type ${dataType} not accepted`); - - const connectorId = String(connector.connector_id); - const addressKey = `${connectorId}|${tag.addr}|${category}`; - let entry = byAddress.get(addressKey); - if (!entry) { - const name = writeTagName(pipeline.schedule_name, legacy, index); - const taken = byName.get(`${connectorId}|${name}`); - if (taken) { - return errors.push(`${where}: name ${name} already used by tag_id ${taken.tag_id} (${taken.address})`); - } - entry = { connector_id: connectorId, address: tag.addr, tag_name: name, category, data_type: dataType, tag_id: null }; - byAddress.set(addressKey, entry); - byName.set(`${connectorId}|${name}`, entry); - newTags.push(entry); - } else if (hasValue(entry.data_type) && entry.data_type !== dataType) { - errors.push(`${where}: tag_id ${entry.tag_id} has data_type ${entry.data_type}, the pipeline ${dataType}`); - } - refs.push({ entry }); - }); - updates.push({ pipeline, refs }); - } - return { errors, newTags, updates, dataTypes }; -}; - -const reserveIds = (count) => { - if (!db.connector_id_counters.findOne({ _id: 'connector_tags' })) { - const max = db.connector_tags.find({}, { tag_id: 1 }).sort({ tag_id: -1 }).limit(1).toArray()[0]; - db.connector_id_counters.insertOne({ _id: 'connector_tags', seq: max ? Number(max.tag_id) : 0 }); - } - const doc = db.connector_id_counters.findOneAndUpdate( - { _id: 'connector_tags' }, - { $inc: { seq: count } }, - { returnDocument: 'after' } - ); - return Number(doc.seq) - count + 1; -}; - -const backfill = () => { - const plan = buildPlan(); - print(`Pipelines to convert: ${plan.updates.length}; connector_tags to create: ${plan.newTags.length}`); - plan.newTags.forEach((t) => print(` + ${t.connector_id} ${t.category} ${t.address} as ${t.tag_name} (${t.data_type})`)); - plan.dataTypes.forEach((dataType, tagId) => print(` * connector_tags ${tagId}: data_type ${dataType}`)); - plan.updates.forEach(({ pipeline, refs }) => - print(` ~ ${pipeline.schedule_name}: ${refs.length} write tag(s)`) - ); - if (plan.errors.length > 0) { - print(`Aborted, nothing was written:\n- ${plan.errors.join('\n- ')}`); - return; - } - if (DRY_RUN) { - print('DRY_RUN: nothing was written (set DRY_RUN = false to apply)'); - return; - } - - const now = new Date(); - if (plan.newTags.length > 0) { - let next = reserveIds(plan.newTags.length); - plan.newTags.forEach((tag) => { - tag.tag_id = next; - next += 1; - }); - db.connector_tags.insertMany( - plan.newTags.map((tag) => ({ - tag_id: tag.tag_id, - connector_id: tag.connector_id, - address: tag.address, - tag_name: tag.tag_name, - category: tag.category, - data_type: tag.data_type, - unit: '', - status: 'Active', - description: '', - metadata: {}, - source: SOURCE, - created_at: now, - updated_at: now, - })) - ); - } - - plan.dataTypes.forEach((dataType, tagId) => { - db.connector_tags.updateOne( - { tag_id: Number(tagId), data_type: { $in: [null, ''] } }, - { $set: { data_type: dataType, updated_at: now } } - ); - }); - - let changed = 0; - const raced = []; - for (const { pipeline, refs } of plan.updates) { - const writeTags = refs.map((ref) => ({ tag_id: ref.tag_id ?? ref.entry.tag_id })); - // Only if write_tags is still the array read above: a pipeline edited meanwhile is skipped - const result = db.pipelines.updateOne( - { _id: pipeline._id, write_tags: pipeline.write_tags }, - { $set: { write_tags: writeTags, legacy_write_tags: pipeline.write_tags, updated_at: now } } - ); - if (result.modifiedCount === 1) changed += 1; - else raced.push(pipeline.schedule_name); - } - print(`connector_tags created: ${plan.newTags.length}; pipelines converted: ${changed}`); - if (raced.length > 0) print(`Changed during the backfill, run it again: ${raced.join(', ')}`); -}; - -const revert = () => { - const pipelines = db.pipelines.find({ legacy_write_tags: { $exists: true } }).toArray(); - // A created tag still referenced by a pipeline that is not restored (cleaned up, or - // pointed at it later) is kept - const inUse = new Set( - db.pipelines - .find({ legacy_write_tags: { $exists: false } }, { read_tags: 1, write_tags: 1 }) - .toArray() - .flatMap((p) => [...(Array.isArray(p.read_tags) ? p.read_tags : []), ...(Array.isArray(p.write_tags) ? p.write_tags : [])]) - .filter((tag) => hasValue(tag?.tag_id)) - .map((tag) => String(tag.tag_id)) - ); - const created = db.connector_tags.find({ source: SOURCE }, { tag_id: 1 }).toArray(); - const toDelete = created.filter((tag) => !inUse.has(String(tag.tag_id))).map((tag) => tag.tag_id); - print( - `Pipelines to restore: ${pipelines.length}; connector_tags to delete: ${toDelete.length}` + - `; kept (still referenced): ${created.length - toDelete.length}` - ); - if (DRY_RUN) { - print('DRY_RUN: nothing was written (set DRY_RUN = false to apply)'); - return; - } - const now = new Date(); - for (const pipeline of pipelines) { - db.pipelines.updateOne( - { _id: pipeline._id }, - { $set: { write_tags: pipeline.legacy_write_tags, updated_at: now }, $unset: { legacy_write_tags: '' } } - ); - } - const { deletedCount } = db.connector_tags.deleteMany({ source: SOURCE, tag_id: { $in: toDelete } }); - print(`pipelines restored: ${pipelines.length}; connector_tags deleted: ${deletedCount}`); -}; - -// Run once the new write tags are confirmed: drops the copy that revert restores from -const cleanup = () => { - const filter = { legacy_write_tags: { $exists: true } }; - const count = db.pipelines.countDocuments(filter); - print(`Pipelines with legacy_write_tags to clean: ${count}`); - if (DRY_RUN) { - print('DRY_RUN: nothing was written (set DRY_RUN = false to apply)'); - return; - } - // updated_at is not bumped: the Orchestrator does not read legacy_write_tags - const { modifiedCount } = db.pipelines.updateMany(filter, { $unset: { legacy_write_tags: '' } }); - print(`legacy_write_tags removed from ${modifiedCount} pipeline(s)`); -}; - -const actions = { backfill, revert, cleanup }; -if (!actions[ACTION]) { - print(`Unknown ACTION ${ACTION}: use backfill, revert or cleanup`); - return; -} -print(`${ACTION}${DRY_RUN ? ' (DRY_RUN)' : ''} on ${DB_NAME}`); -actions[ACTION](); -})();