Processes migrations from a given folder.

  This function uses LOCK TABLE MySQL command on _migrations
  table to ensure that only one GRR process is actually
  performing the migration.

  We have to use open_conn_fn to open 2 connections to the database,
  since LOCK TABLE command is per-connection and it's not allowed
  to modify non-locked tables if LOCK TABLE was called within a
  connection. To overcome this limitation we use one connection
  to lock _migrations and perform its updates and one connection
  to do actual migrations.

  Args:
    open_conn_fn: A function to open new database connection.
    migrations_root: A path to folder with migration files.

def ProcessMigrations(open_conn_fn,
                      migrations_root):
  """Processes migrations from a given folder.

  This function uses LOCK TABLE MySQL command on _migrations
  table to ensure that only one GRR process is actually
  performing the migration.

  We have to use open_conn_fn to open 2 connections to the database,
  since LOCK TABLE command is per-connection and it's not allowed
  to modify non-locked tables if LOCK TABLE was called within a
  connection. To overcome this limitation we use one connection
  to lock _migrations and perform its updates and one connection
  to do actual migrations.

  Args:
    open_conn_fn: A function to open new database connection.
    migrations_root: A path to folder with migration files.
  """
  with contextlib.closing(open_conn_fn()) as conn:
    conn.autocommit(True)

    with contextlib.closing(conn.cursor()) as cursor:
      cursor.execute("""CREATE TABLE IF NOT EXISTS _migrations(
        migration_id INT UNSIGNED PRIMARY KEY,
        timestamp TIMESTAMP(6) NOT NULL DEFAULT NOW(6)
        )""")

    with contextlib.closing(conn.cursor()) as cursor:
      cursor.execute('SELECT GET_LOCK("grr_migration", 3600)')

    try:
      with contextlib.closing(conn.cursor()) as cursor:
        current_migration = GetLatestMigrationNumber(cursor)

      to_process = ListMigrationsToProcess(migrations_root, current_migration)
      logging.info("Will execute following DB migrations: %s",
                   ", ".join(to_process))

      for fname in to_process:
        start_time = time.time()
        logging.info("Starting migration %s", fname)

        with open(os.path.join(migrations_root, fname)) as fd:
          sql = fd.read()
          with contextlib.closing(conn.cursor()) as cursor:
            cursor.execute(sql)

        logging.info("Migration %s is done. Took %.2fs", fname,
                     time.time() - start_time)

        # Update _migrations table with the latest migration.
        with contextlib.closing(conn.cursor()) as cursor:
          cursor.execute("INSERT INTO _migrations (migration_id) VALUES (%s)",
                         [_MigrationFilenameToInt(fname)])
    finally:
      with contextlib.closing(conn.cursor()) as cursor:
        cursor.execute('SELECT RELEASE_LOCK("grr_migration")')