Skip to content

Commit

Permalink
improved disaster recovery mechanism
Browse files Browse the repository at this point in the history
  • Loading branch information
AnatolyUss committed Jan 31, 2017
1 parent 46f1848 commit 86d872e
Show file tree
Hide file tree
Showing 6 changed files with 130 additions and 32 deletions.
2 changes: 1 addition & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -61,7 +61,7 @@ from MySQL to PostgreSQL as easy and smooth as possible.</p>
<a href="mailto:[email protected]?subject=NMIG">[email protected]</a></p>

<h3>VERSION</h3>
<p>Current version is 2.4.0<br />
<p>Current version is 2.5.0<br />
(major version . improvements . bug fixes)</p>


Expand Down
116 changes: 108 additions & 8 deletions migration/fmtp/ConsistencyEnforcer.js
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,9 @@
*/
'use strict';

const generateError = require('./ErrorGenerator');
const extraConfigProcessor = require('./ExtraConfigProcessor');

/**
* Update consistency state.
*
Expand Down Expand Up @@ -53,14 +56,15 @@ const updateConsistencyState = (self, dataPoolId) => {
}

/**
* Get consistency state.
* Get the `is_started` value of current chunk.
*
* @param {Conversion} self
* @param {Number} dataPoolId
*
* @returns {Promise}
*/
const getConsistencyState = (self, dataPoolId) => {

const getIsStarted = (self, dataPoolId) => {
return new Promise(resolve => {
self._pg.connect((error, client, done) => {
if (error) {
Expand All @@ -85,27 +89,123 @@ const getConsistencyState = (self, dataPoolId) => {
});
}

/**
* Current data chunk runs after a disaster recovery.
* Must determine if current chunk has already been loaded.
* This is in order to prevent possible data duplications.
*
* @param {Conversion} self
* @param {Object} chunk
*
* @returns {Promise}
*/
const hasCurrentChunkLoaded = (self, chunk) => {
return new Promise(resolve => {
self._pg.connect((pgError, client, done) => {
if (pgError) {
generateError(self, '\t--[ConsistencyEnforcer::hasCurrentChunkLoaded] Cannot connect to PostgreSQL server...\n' + pgError);
resolve(true);
} else {
const originalTableName = extraConfigProcessor.getTableName(self, chunk._tableName, true);
const sql = 'SELECT EXISTS(SELECT 1 FROM "' + self._schema + '"."' + chunk._tableName
+ '" WHERE "' + self._schema + '_' + originalTableName + '_data_chunk_id_temp" = ' + chunk._id + ');';

client.query(sql, (err, result) => {
done();

if (err) {
generateError(self, '\t--[ConsistencyEnforcer::hasCurrentChunkLoaded] ' + err, sql);
resolve(true);
} else {
resolve(!!result.rows[0].exists);
}
});
}
});
});
}

/**
* Get consistency state.
*
* @param {Conversion} self
* @param {Object} chunk
*
* @returns {Promise}
*/
const getConsistencyState = (self, chunk) => {
return new Promise(resolve => {
getIsStarted(self, chunk._id).then(isStarted => {
if (isStarted) {
hasCurrentChunkLoaded(self, chunk).then(result => resolve(result));
} else {
// Normal migration flow.
resolve(false);
}
});
});
}

/**
* Enforce consistency before processing a chunk of data.
* Ensure there are no any data duplications.
* In case of normal execution - it is a good practice.
* In case of rerunning nmig after unexpected failure - it is absolutely mandatory.
*
* @param {Conversion} self
* @param {Number} chunkId
* @param {Object} chunk
*
* @returns {Promise}
*/
module.exports = (self, chunkId) => {
module.exports.enforceConsistency = (self, chunk) => {
return new Promise(resolve => {
getConsistencyState(self, chunkId).then(isStarted => {
if (isStarted) {
// Current data chunk runs after a disaster recovery.
getConsistencyState(self, chunk).then(hasAlreadyBeenLoaded => {
if (hasAlreadyBeenLoaded) {
/*
* Current data chunk runs after a disaster recovery.
* It has already been loaded.
*/
resolve(false);
} else {
// Normal migration flow.
updateConsistencyState(self, chunkId).then(() => resolve(true));
updateConsistencyState(self, chunk._id).then(() => resolve(true));
}
})
});
};

/**
* Drop the {self._schema + '_' + originalTableName + '_data_chunk_id_temp'} column from current table.
*
* @param {Conversion} self
* @param {String} tableName
*
* @returns {Promise}
*/
module.exports.dropDataChunkIdColumn = (self, tableName) => {
return new Promise(resolve => {
self._pg.connect((pgError, client, done) => {
if (pgError) {
generateError(self, '\t--[ConsistencyEnforcer::dropDataChunkIdColumn] Cannot connect to PostgreSQL server...\n' + pgError);
resolve();
} else {
const originalTableName = extraConfigProcessor.getTableName(self, tableName, true);
const columnToDrop = self._schema + '_' + originalTableName + '_data_chunk_id_temp';
const sql = 'ALTER TABLE "' + self._schema + '"."' + tableName + '" DROP COLUMN "' + columnToDrop + '";';

client.query(sql, (err, result) => {
done();

if (err) {
const errMsg = '\t--[ConsistencyEnforcer::dropDataChunkIdColumn] Failed to drop column "' + columnToDrop + '"\n'
+ '\t--[ConsistencyEnforcer::dropDataChunkIdColumn] '+ err;

generateError(self, errMsg, sql);
}

resolve();
});
}
});
});
};
12 changes: 10 additions & 2 deletions migration/fmtp/ConstraintsProcessor.js
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,8 @@ const processIndexAndKey = require('./IndexAndKeyProcessor');
const processComments = require('./CommentsProcessor');
const processForeignKey = require('./ForeignKeyProcessor');
const processViews = require('./ViewGenerator');
const consistencyEnforcer = require('./ConsistencyEnforcer');
const dropDataChunkIdColumn = consistencyEnforcer.dropDataChunkIdColumn;

/**
* Continues migration process after data loading, when migrate_only_data is true.
Expand All @@ -46,7 +48,11 @@ const continueProcessAfterDataLoadingShort = self => {

for (let i = 0; i < self._tablesToMigrate.length; ++i) {
const tableName = self._tablesToMigrate[i];
promises.push(sequencesProcessor.setSequenceValue(self, tableName));
promises.push(
dropDataChunkIdColumn(self, tableName).then(() => {
return sequencesProcessor.setSequenceValue(self, tableName);
})
);
}

Promise.all(promises).then(() => {
Expand Down Expand Up @@ -77,7 +83,9 @@ const continueProcessAfterDataLoadingLong = self => {
for (let i = 0; i < self._tablesToMigrate.length; ++i) {
const tableName = self._tablesToMigrate[i];
promises.push(
processEnum(self, tableName).then(() => {
dropDataChunkIdColumn(self, tableName).then(() => {
return processEnum(self, tableName);
}).then(() => {
return processNull(self, tableName);
}).then(() => {
return processDefault(self, tableName);
Expand Down
26 changes: 8 additions & 18 deletions migration/fmtp/DataLoader.js
Original file line number Diff line number Diff line change
Expand Up @@ -29,7 +29,8 @@ const generateError = require('./ErrorGenerator');
const connect = require('./Connector');
const Conversion = require('./Conversion');
const MessageToMaster = require('./MessageToMaster');
const enforceConsistency = require('./ConsistencyEnforcer');
const consistencyEnforcer = require('./ConsistencyEnforcer');
const enforceConsistency = consistencyEnforcer.enforceConsistency;
const extraConfigProcessor = require('./ExtraConfigProcessor');
const copyFrom = pgCopyStreams.from;
const getBuffer = +process.version.split('.')[0].slice(1) < 6
Expand All @@ -44,7 +45,7 @@ process.on('message', signal => {
for (let i = 0; i < signal.chunks.length; ++i) {
promises.push(
connect(self).then(() => {
return enforceConsistency(self, signal.chunks[i]._id);
return enforceConsistency(self, signal.chunks[i]);
}).then(isNormalFlow => {
if (isNormalFlow) {
return populateTableWorker(
Expand All @@ -58,19 +59,6 @@ process.on('message', signal => {
);
}

const sql = buildChunkQuery(
extraConfigProcessor.getTableName(self, signal.chunks[i]._tableName, true),
signal.chunks[i]._selectFieldList,
signal.chunks[i]._offset,
signal.chunks[i]._rowsInChunk
);

const strTwelveSpaces = ' ';
const rejectedData = '\n\t--[loadData] Possible data duplication alert!\n\t ' + strTwelveSpaces
+ 'Data, retrievable by following MySQL query:\n' + sql + '\n\t ' + strTwelveSpaces
+ 'may already be migrated.\n\t' + strTwelveSpaces + ' Please, check it.';

log(self, rejectedData, path.join(self._logsDirPath, signal.chunks[i]._tableName + '.log'));
return deleteChunk(self, signal.chunks[i]._id);
})
);
Expand Down Expand Up @@ -225,8 +213,9 @@ const populateTableWorker = (self, tableName, strSelectFieldList, offset, rowsIn
generateError(self, '\t--[populateTableWorker] Cannot connect to MySQL server...\n\t' + error);
resolvePopulateTableWorker();
} else {
const csvAddr = path.join(self._tempDirPath, tableName + offset + '.csv');
const sql = buildChunkQuery(extraConfigProcessor.getTableName(self, tableName, true), strSelectFieldList, offset, rowsInChunk);
const csvAddr = path.join(self._tempDirPath, tableName + offset + '.csv');
const originalTableName = extraConfigProcessor.getTableName(self, tableName, true);
const sql = buildChunkQuery(originalTableName, strSelectFieldList, offset, rowsInChunk);

connection.query(sql, (err, rows) => {
connection.release();
Expand All @@ -235,7 +224,8 @@ const populateTableWorker = (self, tableName, strSelectFieldList, offset, rowsIn
generateError(self, '\t--[populateTableWorker] ' + err, sql);
resolvePopulateTableWorker();
} else {
rowsInChunk = rows.length;
rowsInChunk = rows.length;
rows[0][self._schema + '_' + originalTableName + '_data_chunk_id_temp'] = dataPoolId;

csvStringify(rows, (csvError, csvString) => {
rows = null;
Expand Down
4 changes: 2 additions & 2 deletions migration/fmtp/TableProcessor.js
Original file line number Diff line number Diff line change
Expand Up @@ -126,8 +126,8 @@ module.exports.createTable = (self, tableName) => {
+ '" ' + mapDataTypes(self._dataTypesMap, rows[i].Type) + ',';
}

rows = null;
sql = sql.slice(0, -1) + ');';
sql += '"' + self._schema + '_' + originalTableName + '_data_chunk_id_temp" BIGINT);';

client.query(sql, err => {
done();

Expand Down
2 changes: 1 addition & 1 deletion package.json
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
{
"name": "nmig",
"version": "2.4.0",
"version": "2.5.0",
"description": "The database migration app",
"author": "Anatoly Khaytovich<[email protected]>",
"dependencies": {
Expand Down

0 comments on commit 86d872e

Please sign in to comment.