database/mariadb-columnstore-engine
1
0
Fork 0
mirror of https://github.com/mariadb-corporation/mariadb-columnstore-engine.git synced 2025-11-08 03:42:19 +03:00
Code Activity

MCOL-5555 Add support for startreadonly command. (#3081)

Browse Source 
This patch adds support for `startreadonly` command which waits
until all active cpimport jobs are done and then puts controller node to readonly
mode.
This commit is contained in:
Denis Khalikov
2023-12-26 14:11:33 +03:00
committed by drrtuy
parent 5e2e3d62ff
commit 9caa70988c
11 changed files with 404 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
writeengine/bulk/cpimport.cpp
View File

@@ -58,6 +58,7 @@ namespace
char* pgmName = 0;
const std::string IMPORT_PATH_CWD(".");
bool bDebug = false;
uint32_t cpimportJobId = 0;
//@bug 4643: cpimport job ended during setup w/o any err msg.
// Added a try/catch with logging to main() in case
@@ -191,6 +192,7 @@ void printUsage()
//------------------------------------------------------------------------------
void handleSigTerm(int i)
{
BRMWrapper::getInstance()->finishCpimportJob(cpimportJobId);
std::cout << "Received SIGTERM to terminate the process..." << std::endl;
BulkStatus::setJobStatus(EXIT_FAILURE);
}
@@ -200,12 +202,32 @@ void handleSigTerm(int i)
//------------------------------------------------------------------------------
void handleControlC(int i)
{
BRMWrapper::getInstance()->finishCpimportJob(cpimportJobId);
if (!BulkLoad::disableConsoleOutput())
std::cout << "Received Control-C to terminate the process..." << std::endl;
BulkStatus::setJobStatus(EXIT_FAILURE);
}
//------------------------------------------------------------------------------
// Signal handler to catch SIGTERM signal to terminate the process
//------------------------------------------------------------------------------
void handleSigSegv(int i)
{
BRMWrapper::getInstance()->finishCpimportJob(cpimportJobId);
std::cout << "Received SIGSEGV to terminate the process..." << std::endl;
BulkStatus::setJobStatus(EXIT_FAILURE);
}
//------------------------------------------------------------------------------
// Signal handler to catch SIGTERM signal to terminate the process
//------------------------------------------------------------------------------
void handleSigAbrt(int i)
{
BRMWrapper::getInstance()->finishCpimportJob(cpimportJobId);
std::cout << "Received SIGABRT to terminate the process..." << std::endl;
BulkStatus::setJobStatus(EXIT_FAILURE);
}
//------------------------------------------------------------------------------
// If error occurs during startup, this function is called to log the specified
@@ -213,6 +235,7 @@ void handleControlC(int i)
//------------------------------------------------------------------------------
void startupError(const std::string& errMsg, bool showHint)
{
BRMWrapper::getInstance()->finishCpimportJob(cpimportJobId);
// Log to console
if (!BulkLoad::disableConsoleOutput())
cerr << errMsg << endl;
@@ -274,6 +297,17 @@ void setupSignalHandlers()
memset(&act, 0, sizeof(act));
act.sa_handler = handleSigTerm;
sigaction(SIGTERM, &act, 0);
// catch SIGSEGV signal to terminate the program
memset(&act, 0, sizeof(act));
act.sa_handler = handleSigSegv;
sigaction(SIGSEGV, &act, 0);
// catch SIGABRT signal to terminate the program
memset(&act, 0, sizeof(act));
act.sa_handler = handleSigAbrt;
sigaction(SIGABRT, &act, 0);
}
//------------------------------------------------------------------------------
@@ -1266,6 +1300,16 @@ int main(int argc, char** argv)
new boost::thread(utils::MonitorProcMem(0, checkPct, SUBSYSTEM_ID_WE_BULK));
}
rc = BRMWrapper::getInstance()->newCpimportJob(cpimportJobId);
if (rc != NO_ERROR)
{
WErrorCodes ec;
std::ostringstream oss;
oss << "Error in creating new cpimport job on Controller node; " << ec.errorString(rc)
<< "; cpimport is terminating.";
startupError(oss.str(), false);
}
//--------------------------------------------------------------------------
// This is the real business
//--------------------------------------------------------------------------
@@ -1316,6 +1360,8 @@ int main(int argc, char** argv)
rc = ERR_UNKNOWN;
}
BRMWrapper::getInstance()->finishCpimportJob(cpimportJobId);
//--------------------------------------------------------------------------
// Log end of job to INFO log
//--------------------------------------------------------------------------