Commit 7da170f2 authored by Michael Davis's avatar Michael Davis
Browse files

[migration] Creates top-level directory for import if it doesn't already exist

parent 05df2c39
......@@ -83,8 +83,10 @@ EosImportDirs::EosImportDirs(const std::string &configfile, std::string topLevel
}
void EosImportDirs::select()
void EosImportDirs::select(std::string tld)
{
if(tld.empty()) tld = m_topLevelPath + '%';
const std::string sql =
"SELECT "
" FILEID,"
......@@ -102,7 +104,7 @@ void EosImportDirs::select()
"ORDER BY "
" DEPTH ASC";
m_selectDirsDbStmt = m_dbConn.createStmt(sql);
m_selectDirsDbStmt.bindString(":TOP_LEVEL_PATH", m_topLevelPath + "%");
m_selectDirsDbStmt.bindString(":TOP_LEVEL_PATH", tld);
std::cerr << "Executing SELECT query against " << m_tableName << " table...";
m_selectDirsDbRset = m_selectDirsDbStmt.executeQuery();
std::cerr << "done." << std::endl;
......@@ -234,13 +236,13 @@ bool EosImportDirs::compareMD(const eos::rpc::ContainerMdProto &dir)
auto remote_dir = m_eosgrpc->GetMD(eos::rpc::CONTAINER, dir.id(), dir.path());
// Check directory metadata is the same
if(dir.id() != remote_dir.cmd().id() ||
dir.uid() != remote_dir.cmd().uid() ||
dir.gid() != remote_dir.cmd().gid() ||
dir.mode() != remote_dir.cmd().mode() ||
dir.ctime().sec() != remote_dir.cmd().ctime().sec() ||
dir.name() != remote_dir.cmd().name() ||
(dir.path() + '/') != remote_dir.cmd().path()) {
if(dir.id() != remote_dir.cmd().id() ||
dir.uid() != remote_dir.cmd().uid() ||
dir.gid() != remote_dir.cmd().gid() ||
dir.mode() != remote_dir.cmd().mode() ||
dir.ctime().sec() != remote_dir.cmd().ctime().sec() ||
dir.name() != remote_dir.cmd().name() ||
(dir.path() + '/') != remote_dir.cmd().path()) {
return false;
}
......@@ -300,6 +302,31 @@ void EosImportDirs::forgetFailures()
}
void EosImportDirs::checkRootDirExists()
{
auto tld = eos::client::manglePathname(m_castor_prefix, m_eos_prefix, m_topLevelPath);
std::string &path = tld.pathname;
auto tld_pb = m_eosgrpc->GetMD(eos::rpc::CONTAINER, 0, path);
// If the root directory does not exist, try to create the tree up to the root
std::list<std::string> dirsToCreate;
while(tld_pb.cmd().id() == 0) {
// Strip off any trailing slashes
while(!path.empty() && path[path.length()-1] == '/') {
path.resize(path.length()-1);
}
dirsToCreate.push_front(path);
path.resize(path.find_last_of('/'));
tld_pb = m_eosgrpc->GetMD(eos::rpc::CONTAINER, 0, path);
}
for(auto &dir : dirsToCreate) {
std::cerr << "Creating " << dir << ":" << std::endl;
select(dir);
processBatch();
}
}
void EosImportDirs::getStorageClasses()
{
std::cerr << "Reading storage classes from CASTOR DB...";
......
......@@ -28,11 +28,12 @@ class EosImportDirs
public:
EosImportDirs(const std::string &configfile, std::string topLevelPath = "/", bool delta = false, bool retry = false);
void checkRootDirExists();
void getStorageClasses();
void listFailureModes();
void listFailures();
void forgetFailures();
void select();
void select(std::string tld = "");
bool processBatch();
private:
......
......@@ -205,6 +205,7 @@ int main(const int argc, char ** argv)
case EosImportDirsCmdLine::INJECT: {
cta::migration::EosImportDirs importDirs(cmdLine.configPath, cmdLine.topLevelPath, cmdLine.delta, cmdLine.retry);
importDirs.getStorageClasses();
importDirs.checkRootDirExists();
importDirs.select();
while(importDirs.processBatch());
}
......
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment