TRX/Checkmate
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #3190 from bluewave-labs/fix/migration

Browse Source 
fix: migration
This commit is contained in:
Alexander Holliday authored and GitHub committed 2026-01-26 15:44:50 -08:00
2 parents e5bb736530 716479c5bb
commit ba498534e6
3 files changed
+148 -59
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

server/src/db/migration/0002_convertChecksToTimeSeries.ts
+133 -44
View File
@@ -1,9 +1,17 @@
import mongoose from "mongoose";
import { BulkWriteResult } from "mongodb";
const CHECKS_COLLECTION = "checks";
const BACKUP_COLLECTION = "checks_backup";
const FAILED_DOCS_COLLECTION = "checks_migration_failed";
const BATCH_SIZE = 1000;
interface MigrationStats {
totalSource: number;
totalMigrated: number;
totalFailed: number;
}
const getDb = () => {
const db = mongoose.connection.db;
if (!db) {
@@ -12,12 +20,31 @@ const getDb = () => {
return db;
};
const restoreFromBackup = async () => {
const db = getDb();
const backupExists = await db.listCollections({ name: BACKUP_COLLECTION }).toArray();
if (backupExists.length === 0) {
throw new Error("Cannot restore: backup collection does not exist");
}
const checksExists = await db.listCollections({ name: CHECKS_COLLECTION }).toArray();
if (checksExists.length > 0) {
await db.collection(CHECKS_COLLECTION).drop();
}
await db.collection(BACKUP_COLLECTION).rename(CHECKS_COLLECTION);
};
const backupAndDropExistingCollection = async () => {
const db = getDb();
const collections = await db.listCollections({ name: CHECKS_COLLECTION }).toArray();
if (collections.length === 0) {
return false;
}
const backupExists = await db.listCollections({ name: BACKUP_COLLECTION }).toArray();
if (backupExists.length > 0) {
throw new Error(`Backup collection "${BACKUP_COLLECTION}" already exists. ` + `Please remove it manually before running migration.`);
}
await db
.collection(CHECKS_COLLECTION)
@@ -27,69 +54,131 @@ const backupAndDropExistingCollection = async () => {
return true;
};
const createTimeSeriesCollection = async () => {
const createTimeSeriesCollection = async (backedUp: boolean) => {
const db = getDb();
const existing = await db.listCollections({ name: CHECKS_COLLECTION }).toArray();
if (existing.length === 0) {
await db.createCollection(CHECKS_COLLECTION, {
timeseries: {
timeField: "createdAt",
metaField: "metadata",
granularity: "seconds",
},
});
}
try {
const existing = await db.listCollections({ name: CHECKS_COLLECTION }).toArray();
if (existing.length === 0) {
await db.createCollection(CHECKS_COLLECTION, {
timeseries: {
timeField: "createdAt",
metaField: "metadata",
granularity: "seconds",
},
});
}
await db
.collection(CHECKS_COLLECTION)
.createIndexes([
{ key: { updatedAt: 1 } },
{ key: { "metadata.monitorId": 1, updatedAt: 1 } },
{ key: { "metadata.monitorId": 1, updatedAt: -1 } },
{ key: { "metadata.teamId": 1, updatedAt: -1 } },
]);
await db
.collection(CHECKS_COLLECTION)
.createIndexes([
{ key: { createdAt: 1 } },
{ key: { "metadata.monitorId": 1, createdAt: 1 } },
{ key: { "metadata.monitorId": 1, createdAt: -1 } },
{ key: { "metadata.teamId": 1, createdAt: -1 } },
{ key: { "metadata.monitorId": 1, "metadata.type": 1, createdAt: -1 } },
{ key: { "metadata.teamId": 1, status: 1, createdAt: -1 } },
]);
} catch (error) {
if (backedUp) {
await restoreFromBackup();
}
throw error;
}
};
const migrateBackupData = async () => {
const validateDocument = (doc: any): boolean => {
if (!doc.createdAt) {
return false;
}
if (!doc.monitorId || !doc.teamId || !doc.type) {
return false;
}
return true;
};
const storeFailedDocument = async (doc: any, reason: string) => {
const db = getDb();
await db.collection(FAILED_DOCS_COLLECTION).insertOne({
originalDoc: doc,
reason,
failedAt: new Date(),
});
};
const migrateBackupData = async (backedUp: boolean): Promise<MigrationStats> => {
const db = getDb();
const backupExists = await db.listCollections({ name: BACKUP_COLLECTION }).toArray();
if (backupExists.length === 0) {
return;
return { totalSource: 0, totalMigrated: 0, totalFailed: 0 };
}
const source = db.collection(BACKUP_COLLECTION);
const target = db.collection(CHECKS_COLLECTION);
const cursor = source.find();
const operations: any[] = [];
const stats: MigrationStats = {
totalSource: 0,
totalMigrated: 0,
totalFailed: 0,
};
while (await cursor.hasNext()) {
const doc = await cursor.next();
if (!doc) {
continue;
try {
const source = db.collection(BACKUP_COLLECTION);
const target = db.collection(CHECKS_COLLECTION);
stats.totalSource = await source.countDocuments();
const cursor = source.find().addCursorFlag("noCursorTimeout", true);
const operations: any[] = [];
const invalidDocs: any[] = [];
while (await cursor.hasNext()) {
const doc = await cursor.next();
if (!doc) {
continue;
}
if (!validateDocument(doc)) {
invalidDocs.push(doc);
continue;
}
const { _id, monitorId, teamId, type, ...rest } = doc;
const metadata = { monitorId, teamId, type };
operations.push({ insertOne: { document: { ...rest, metadata } } });
if (operations.length >= BATCH_SIZE) {
const result: BulkWriteResult = await target.bulkWrite(operations, { ordered: false });
stats.totalMigrated += result.insertedCount;
operations.length = 0;
}
}
const metadata = {
monitorId: doc.monitorId,
teamId: doc.teamId,
type: doc.type,
};
operations.push({ insertOne: { document: { ...doc, metadata } } });
if (operations.length >= BATCH_SIZE) {
await target.bulkWrite(operations);
operations.length = 0;
if (operations.length) {
const result: BulkWriteResult = await target.bulkWrite(operations, { ordered: false });
stats.totalMigrated += result.insertedCount;
}
}
if (operations.length) {
await target.bulkWrite(operations);
for (const doc of invalidDocs) {
await storeFailedDocument(doc, "Missing required fields (createdAt, monitorId, teamId, or type)");
stats.totalFailed++;
}
await cursor.close();
return stats;
} catch (error) {
if (backedUp) {
await restoreFromBackup();
}
throw error;
}
};
export const convertChecksToTimeSeries = async () => {
const backedUp = await backupAndDropExistingCollection();
await createTimeSeriesCollection();
await createTimeSeriesCollection(backedUp);
if (backedUp) {
await migrateBackupData();
const stats = await migrateBackupData(backedUp);
console.log(
`Migration completed: ${stats.totalMigrated} documents migrated, ` +
`${stats.totalFailed} documents failed (stored in ${FAILED_DOCS_COLLECTION})`
);
}
};
server/src/db/models/Check.ts
+4 -4
View File
@@ -284,12 +284,12 @@ const CheckSchema = new Schema<CheckDocument>(
}
);
CheckSchema.index({ updatedAt: 1 });
CheckSchema.index({ "metadata.monitorId": 1, updatedAt: 1 });
CheckSchema.index({ "metadata.monitorId": 1, updatedAt: -1 });
CheckSchema.index({ "metadata.monitorId": 1, createdAt: -1 });
CheckSchema.index({ "metadata.teamId": 1, updatedAt: -1 });
CheckSchema.index({ "metadata.monitorId": 1, createdAt: 1 });
CheckSchema.index({ createdAt: 1 });
CheckSchema.index({ "metadata.teamId": 1, createdAt: -1 });
CheckSchema.index({ "metadata.monitorId": 1, "metadata.type": 1, createdAt: -1 });
CheckSchema.index({ "metadata.teamId": 1, status: 1, createdAt: -1 });
const CheckModel = model<CheckDocument>("Check", CheckSchema);
server/src/repositories/checks/MongoChecksRepistory.ts
+11 -11
View File
@@ -440,11 +440,11 @@ class MongoChecksRepository implements IChecksRepository {
) => {
const matchStage = {
"metadata.monitorId": monitorObjectId,
updatedAt: { $gte: startDate, $lte: endDate },
createdAt: { $gte: startDate, $lte: endDate },
};
const [result] = await CheckModel.aggregate([
{ $match: matchStage },
{ $sort: { updatedAt: 1 } },
{ $sort: { createdAt: 1 } },
{
$facet: {
uptimePercentage: [
@@ -646,7 +646,7 @@ class MongoChecksRepository implements IChecksRepository {
avgTemperatures: { $push: { $ifNull: ["$cpu.temperature", [0]] } },
disks: { $push: "$disk" },
net: { $push: "$net" },
updatedAts: { $push: "$updatedAt" },
createdAts: { $push: "$createdAt" },
sampleDoc: { $first: "$$ROOT" },
},
},
@@ -685,7 +685,7 @@ class MongoChecksRepository implements IChecksRepository {
bytesSentPerSecond: {
$let: {
vars: {
tDiff: { $divide: [{ $subtract: [{ $last: "$updatedAts" }, { $first: "$updatedAts" }] }, 1000] },
tDiff: { $divide: [{ $subtract: [{ $last: "$createdAts" }, { $first: "$createdAts" }] }, 1000] },
f: { $arrayElemAt: [{ $map: { input: { $first: "$net" }, as: "i", in: "$$i.bytes_sent" } }, "$$nIdx"] },
l: { $arrayElemAt: [{ $map: { input: { $last: "$net" }, as: "i", in: "$$i.bytes_sent" } }, "$$nIdx"] },
},
@@ -695,7 +695,7 @@ class MongoChecksRepository implements IChecksRepository {
deltaBytesRecv: {
$let: {
vars: {
tDiff: { $divide: [{ $subtract: [{ $last: "$updatedAts" }, { $first: "$updatedAts" }] }, 1000] },
tDiff: { $divide: [{ $subtract: [{ $last: "$createdAts" }, { $first: "$createdAts" }] }, 1000] },
f: { $arrayElemAt: [{ $map: { input: { $first: "$net" }, as: "i", in: "$$i.bytes_recv" } }, "$$nIdx"] },
l: { $arrayElemAt: [{ $map: { input: { $last: "$net" }, as: "i", in: "$$i.bytes_recv" } }, "$$nIdx"] },
},
@@ -705,7 +705,7 @@ class MongoChecksRepository implements IChecksRepository {
deltaPacketsSent: {
$let: {
vars: {
tDiff: { $divide: [{ $subtract: [{ $last: "$updatedAts" }, { $first: "$updatedAts" }] }, 1000] },
tDiff: { $divide: [{ $subtract: [{ $last: "$createdAts" }, { $first: "$createdAts" }] }, 1000] },
f: { $arrayElemAt: [{ $map: { input: { $first: "$net" }, as: "i", in: "$$i.packets_sent" } }, "$$nIdx"] },
l: { $arrayElemAt: [{ $map: { input: { $last: "$net" }, as: "i", in: "$$i.packets_sent" } }, "$$nIdx"] },
},
@@ -715,7 +715,7 @@ class MongoChecksRepository implements IChecksRepository {
deltaPacketsRecv: {
$let: {
vars: {
tDiff: { $divide: [{ $subtract: [{ $last: "$updatedAts" }, { $first: "$updatedAts" }] }, 1000] },
tDiff: { $divide: [{ $subtract: [{ $last: "$createdAts" }, { $first: "$createdAts" }] }, 1000] },
f: { $arrayElemAt: [{ $map: { input: { $first: "$net" }, as: "i", in: "$$i.packets_recv" } }, "$$nIdx"] },
l: { $arrayElemAt: [{ $map: { input: { $last: "$net" }, as: "i", in: "$$i.packets_recv" } }, "$$nIdx"] },
},
@@ -725,7 +725,7 @@ class MongoChecksRepository implements IChecksRepository {
deltaErrIn: {
$let: {
vars: {
tDiff: { $divide: [{ $subtract: [{ $last: "$updatedAts" }, { $first: "$updatedAts" }] }, 1000] },
tDiff: { $divide: [{ $subtract: [{ $last: "$createdAts" }, { $first: "$createdAts" }] }, 1000] },
f: { $arrayElemAt: [{ $map: { input: { $first: "$net" }, as: "i", in: "$$i.err_in" } }, "$$nIdx"] },
l: { $arrayElemAt: [{ $map: { input: { $last: "$net" }, as: "i", in: "$$i.err_in" } }, "$$nIdx"] },
},
@@ -735,7 +735,7 @@ class MongoChecksRepository implements IChecksRepository {
deltaErrOut: {
$let: {
vars: {
tDiff: { $divide: [{ $subtract: [{ $last: "$updatedAts" }, { $first: "$updatedAts" }] }, 1000] },
tDiff: { $divide: [{ $subtract: [{ $last: "$createdAts" }, { $first: "$createdAts" }] }, 1000] },
f: { $arrayElemAt: [{ $map: { input: { $first: "$net" }, as: "i", in: "$$i.err_out" } }, "$$nIdx"] },
l: { $arrayElemAt: [{ $map: { input: { $last: "$net" }, as: "i", in: "$$i.err_out" } }, "$$nIdx"] },
},
@@ -745,7 +745,7 @@ class MongoChecksRepository implements IChecksRepository {
deltaDropIn: {
$let: {
vars: {
tDiff: { $divide: [{ $subtract: [{ $last: "$updatedAts" }, { $first: "$updatedAts" }] }, 1000] },
tDiff: { $divide: [{ $subtract: [{ $last: "$createdAts" }, { $first: "$createdAts" }] }, 1000] },
f: { $arrayElemAt: [{ $map: { input: { $first: "$net" }, as: "i", in: "$$i.drop_in" } }, "$$nIdx"] },
l: { $arrayElemAt: [{ $map: { input: { $last: "$net" }, as: "i", in: "$$i.drop_in" } }, "$$nIdx"] },
},
@@ -755,7 +755,7 @@ class MongoChecksRepository implements IChecksRepository {
deltaDropOut: {
$let: {
vars: {
tDiff: { $divide: [{ $subtract: [{ $last: "$updatedAts" }, { $first: "$updatedAts" }] }, 1000] },
tDiff: { $divide: [{ $subtract: [{ $last: "$createdAts" }, { $first: "$createdAts" }] }, 1000] },
f: { $arrayElemAt: [{ $map: { input: { $first: "$net" }, as: "i", in: "$$i.drop_out" } }, "$$nIdx"] },
l: { $arrayElemAt: [{ $map: { input: { $last: "$net" }, as: "i", in: "$$i.drop_out" } }, "$$nIdx"] },
},