SIENTIAPDE-1974
SIENTIAPDE-1974-connectors-pi-web-api: Remove write tag backfill script and update README to require pre-deployment conversion of write tags to [{tag_id}] format.
This commit is contained in:
@@ -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
|
||||
|
||||
|
||||
@@ -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 <schedule>:<type>[: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: <schedule>:<type>, plus :<n> 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]();
|
||||
})();
|
||||
Reference in New Issue
Block a user