DBSync.cc

Go to the documentation of this file.

//-------------------------------------------------------------------------------------------------------------------------------
// MYSQL Database Synchronization Utility
//
// Jared MacLeod jmacleod@lynx.neu.edu
// Tom Paul tom.paul@cern.ch
//
// This program will only work if both the Main Mysql Server and the local one have the same tables setup identically for the database.
//
// INPUT-  Takes command line input.  Must be in correct order.  See Usage below.  Works on one database at a time.
//
// Uses input information to connect to 2 mysql servers with the same database and syncs them using timestamps as reference.
// Note that all tables need a timestamp column in order to be synchronized.
//
//
//-------------------------------------------------------------------------------------------------------------------------------

#include <mysql/mysql.h>
#include <string>
#include <vector>
#include <sstream>
#include <iostream>
#include <iterator>

using namespace std;


void
DisplayUsage(const char* const pname)
{
  cout << "Usage : " << pname << " "
          "-d {database name(s)} -r {remote server to read from} "
          "-U {username on read server} -P {pasword on read server} "
          "-w {local server to write to} -u {username on write server} "
          "-p {password on write server} [-o {port on read server} "
          "-i {port on write server}]\n"
       << endl;

  exit(EXIT_SUCCESS);
}


// we do not want unnecessary dependency, therefore this poor-man's implementation of boost::lexical_cast

template<typename T, typename U>
T
LexicalCast(const U& x)
{
  stringstream ss;
  ss << x;
  T y;
  if (!(ss >> y))
    throw;
  return y;
}


int
main(int argc, char* argv[])
{
  string databaseName;
  string dbserverMain;
  string usernameMain;
  string passwordMain;
  string dbserverLocal;
  string usernameLocal;
  string passwordLocal;
  string localPortString = "3306";
  string remotePortString = "3306";

  if (argc == 1)
    DisplayUsage(argv[0]);

  for (int i = 1; i < argc && *argv[i] == '-'; ++i)
    for (const char* p = &argv[i][1]; *p != '\0'; ++p)
      switch (*p) {
      case 'd':
        if (argv[i+1])
          databaseName = argv[++i];
        break;
      case 'r':
        if (argv[i+1])
          dbserverMain = argv[++i];
        break;
      case 'U':
        if (argv[i+1])
          usernameMain = argv[++i];
        break;
      case 'P':
        if (argv[i+1])
          passwordMain = argv[++i];
        break;
      case 'w':
        if (argv[i+1])
          dbserverLocal = argv[++i];
        break;
      case 'u':
        if (argv[i+1])
          usernameLocal = argv[++i];
        break;
      case 'p':
        if (argv[i+1])
          passwordLocal = argv[++i];
        break;
      case 'o':
        if (argv[i+1])
          remotePortString = argv[++i];
        break;
      case 'i':
        if (argv[i+1])
          localPortString = argv[++i];
        break;
      default:
        break;
      }

  if (databaseName.empty() ||
      dbserverMain.empty() ||
      usernameMain.empty() ||
      passwordMain.empty() ||
      dbserverLocal.empty() ||
      usernameLocal.empty() ||
      passwordLocal.empty()) {
    cout << "Required argument is missing." << endl;
    DisplayUsage(argv[0]);
  }

  // Parse the databaseName string and convert it into multiple strings, each with the name of a single database
  istringstream databaseNameStream(databaseName);
  const vector<string> databaseNames{istream_iterator<string>(databaseNameStream),
                                     istream_iterator<string>()};

  // Some checking should go here to ensure databases are identical before attempting an update
  // This (probably) means checking the primary keys on the last entry in all tables in the database.
  // Is that a sufficient check?

  cout << "\n"
          "=========================================================================================="
       << endl;

  const time_t curTime = time(nullptr);
  cout << ctime(&curTime) << "\n"
          "Synchronizing databases from server "
       << dbserverMain << " "
          "(port " << remotePortString << ") "
          "to server " << dbserverLocal << " "
          "(port " << localPortString << ")\n"
          "The following database(s) will be affected:\n";

  for (const auto& db : databaseNames)
    cout << db << '\n';
  cout << flush;

  //--------------------------------------------------------------------------------------------
  // This section attempts to connect to the databases, and quits if it cannot connect to either.
  //--------------------------------------------------------------------------------------------

  MYSQL* sqlMain = nullptr;
  MYSQL* sqlLocal = nullptr;

  for (const auto& db : databaseNames) {

    cout << "------------------- Updating database: " << db << " -----------------------" << endl;

    sqlMain = mysql_init(nullptr);
    sqlLocal = mysql_init(nullptr);

    const int localPort = LexicalCast<int>(localPortString);
    const int remotePort = LexicalCast<int>(remotePortString);

    if (!mysql_real_connect(sqlMain, dbserverMain.c_str(),
                            usernameMain.c_str(), passwordMain.c_str(),
                            db.c_str(), remotePort, 0, 0)) {
      cerr << sqlMain << ' ' << dbserverMain.c_str() << ' '
           << usernameMain.c_str() << ' '
           << passwordMain.c_str() << ' '
           << db.c_str() << ' '
           << remotePort << "\n"
              "Failed to connect to remote database: "
              "Error: " << mysql_error(sqlMain) << endl;
      mysql_close(sqlMain);
      mysql_close(sqlLocal);
      return 1;
    } else if (!mysql_real_connect(sqlLocal, dbserverLocal.c_str(),
                                   usernameLocal.c_str(), passwordLocal.c_str(),
                                   db.c_str(), localPort, 0, 0)) {
      cerr << "Failed to connect to local database: "
              "Error: " << mysql_error(sqlLocal) << endl;
      mysql_close(sqlMain);
      mysql_close(sqlLocal);
      return 2;
    }

    //-----------------------------------------------------------------------
    //This section gets a list of the tables in the selected database.
    //-----------------------------------------------------------------------

    ostringstream query;
    query << "SHOW TABLES";
    if (mysql_query(sqlMain, query.str().c_str())) {

      cerr << "**Error: Could not get a list of tables from Main server: " << dbserverMain << "\n"
              "  Query failed : " << query.str().c_str() << endl;

    } else {

      MYSQL_ROW row;
      MYSQL_ROW row2;
      MYSQL_RES* databaseRes = nullptr;
      databaseRes = mysql_store_result(sqlMain);
      const int numTables = mysql_num_rows(databaseRes);

      if (!numTables)
        cerr << "*Warning: database contains no tables!" << endl;

      for (int i = 0; i < numTables; ++i) {
        //-------------------------------------------------
        // Find the last primary key number in local table.
        //-------------------------------------------------

        row = mysql_fetch_row(databaseRes);
        const string tableName = row[0];
        query.str("");

        query << "SELECT " << tableName << "_id FROM "
              << tableName << " ORDER BY " << tableName << "_id DESC LIMIT 1";

        if (mysql_query(sqlLocal, query.str().c_str())) {

          cerr << "** Error trying to determine the last modified time for " << tableName << "\n"
                  "  Query failed: " << query.str().c_str() << endl;

        } else {

          MYSQL_RES* const localPrimaryKeyRes = mysql_store_result(sqlLocal);
          const int numLocalRecords = mysql_num_rows(localPrimaryKeyRes);
          int lastLocalPrimaryKey = 0;
          int lastPrimaryKey = 0;

          if (numLocalRecords > 0) {
            const MYSQL_ROW localPrimaryKeyRow = mysql_fetch_row(localPrimaryKeyRes);
            lastLocalPrimaryKey = LexicalCast<int>(localPrimaryKeyRow[0]);
            cout << "Updating table: " << tableName << ". Last primary key for local table = " << lastLocalPrimaryKey << endl;
          } else {
            lastLocalPrimaryKey = 0;
            cout << "No records in local table: " << tableName << ". "
                    "Attempting to fill entire local table." << endl;
          }

          // First check that local database does not have primary keys with greater value than in remote database.
          // If this is the case, it implies that local database was modified using some code other than DBSync

          query.str("");
          query << "SELECT " << tableName << "_id FROM "
                << tableName
            //    << " WHERE " << tableName
            //    << "_id <= " << lastLocalPrimaryKey << " "
                << " ORDER BY " << tableName << "_id DESC LIMIT 1";

          if (!mysql_query(sqlMain, query.str().c_str())) {

            MYSQL_RES* const tableRes = mysql_store_result(sqlMain);
            const int numRows = mysql_num_rows(tableRes);

            if (numRows > 0) {
              MYSQL_ROW firstKeyRow = mysql_fetch_row(tableRes);
              lastPrimaryKey = LexicalCast<int>(firstKeyRow[0]);
            }

            if (numRows < numLocalRecords || lastPrimaryKey < lastLocalPrimaryKey) {
              cout << "MAJOR BIG-TIME ERROR!\n"
                      "The number of records in the local table '" << tableName << "' "
                      "(" << numLocalRecords << ")\n and the number of records in the master table "
                      "(" << numRows << ") are not equal\n"
                      "or the last key in the local table (" << lastLocalPrimaryKey << ") "
                      "and the last key in the master table (" << lastPrimaryKey << ") "
                      "are not equal!\n"
                      "Has someone been writing to the local database?" << endl;
              exit(1);
            }
            mysql_free_result(tableRes);
          }

          const int limit = 1000;
          int numWrites = 0;
          ostringstream insertString;

          for (;;) {

            query.str("");
            query << "SELECT * FROM " << tableName << " WHERE " << tableName << "_id > " << lastLocalPrimaryKey << " AND "
                  << tableName << "_id <= " << lastLocalPrimaryKey + limit << ';';

            int numRows = 0;

            if (mysql_query(sqlMain, query.str().c_str())) {

              cerr << "**Error Could not get records from Main database\n"
                      "  Query failed: " << query.str().c_str() << endl;
              mysql_close(sqlMain);
              mysql_close(sqlLocal);
              exit(3);

            } else {

              MYSQL_RES* const tableRes = mysql_store_result(sqlMain);
              numRows = mysql_num_rows(tableRes);

              if (!numRows)

                cout << "No new records.  Nothing is being written in the database." << endl;

              else {

                const int numFields = mysql_num_fields(tableRes);

                if (!numFields)
                  cerr << "*Warning: zero fields!" << endl;

                cout << "*********** Writing " << ++numWrites << ": " << numRows
                     << " new records in " << tableName
                     << " (limit per query = " << limit << ") " << endl;

                insertString.str("");
                insertString << "INSERT INTO " << tableName << " VALUES ";

                for (int j = 0; j < numRows; ++j) {

                  row2 = mysql_fetch_row(tableRes);
                  insertString << "( ";

                  for (int k = 0; k < numFields; ++k) {

                    if (row2[k])
                      insertString << '"' << row2[k] << '"';
                    else
                      insertString << "NULL";

                    if (k < (numFields - 1))
                      insertString << ", ";

                  }

                  if (j < (numRows - 1))
                    insertString << " ), ";
                  else
                    insertString << " );";

                }

                if (mysql_query(sqlLocal, insertString.str().c_str()))
                  cerr << "**ERROR. Insert query to the database failed.\n"
                          "query : " << insertString.str() << "\n"
                          "reported error: " << mysql_error(sqlLocal) << endl;

              }

              mysql_free_result(tableRes);

            }

            if (lastPrimaryKey <= lastLocalPrimaryKey)
              break;

            lastLocalPrimaryKey += limit;

          }
        }
      }
    }

    mysql_close(sqlMain);
    mysql_close(sqlLocal);

  }
}