* Copyright (c) 2021 Huawei Technologies Co.,Ltd.
*
* CM is licensed under Mulan PSL v2.
* You can use this software according to the terms and conditions of the Mulan PSL v2.
* You may obtain a copy of Mulan PSL v2 at:
*
* http://license.coscl.org.cn/MulanPSL2
*
* THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
* EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
* MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
* See the Mulan PSL v2 for more details.
* -------------------------------------------------------------------------
*
* cma_create_conn_cms.cpp
*
*
* IDENTIFICATION
* src/cm_agent/cma_create_conn_cms.cpp
*
* -------------------------------------------------------------------------
*/
#include "cma_global_params.h"
#include "cma_common.h"
#include "cma_connect.h"
#include "cma_create_conn_cms.h"
#define DISABLE_TIMEOUT 0
#define NOT_UPGRADED 0
#define GRAYSCALE_UPGRADED 1
#define INPLACE_UPGRADED 2
* This flag is activated after the cm agent is connected to the cm Server for the first time.
* If this flag is not activated, we will disable some cm agent features.
*/
bool g_firstConnectFlag = false;
bool g_agentConnCmsSuccess = false;
int CreateStopNodeInstancesFlagFile(const char *stopFlagFile)
{
char cmd[MAX_PATH_LEN] = {0};
int ret = snprintf_s(cmd, MAX_PATH_LEN, MAX_PATH_LEN - 1, "touch %s;chmod 600 %s", stopFlagFile, stopFlagFile);
securec_check_intval(ret, (void)ret);
ret = system(cmd);
if (ret != 0) {
write_runlog(ERROR, "CreateStopNodeInstancesFlagFile failed:%s, errnum=%d, errno=%d.\n", cmd, ret, errno);
return -1;
}
write_runlog(LOG, "CreateStopNodeInstancesFlagFile success: %s.\n", cmd);
return 0;
}
void StartOrStopInstanceByCommand(OperateType operateType, uint32 nodeId, const char *instanceDataPath)
{
errno_t rc = 0;
char cmd[MAXPGPATH] = {0};
if (operateType == INSTANCE_START) {
rc = snprintf_s(cmd, MAXPGPATH, MAXPGPATH - 1,
"cm_ctl start -n %u -D %s > /dev/null 2>&1 &", nodeId, instanceDataPath);
} else if (operateType == INSTANCE_STOP) {
rc = snprintf_s(cmd, MAXPGPATH, MAXPGPATH - 1,
"cm_ctl stop -n %u -D %s -m i > /dev/null 2>&1 &", nodeId, instanceDataPath);
} else {
write_runlog(ERROR, "StartOrStopInstanceByCommand invaild operateType:%d\n", operateType);
return;
}
securec_check_intval(rc, (void)rc);
int ret = system(cmd);
if (ret != 0) {
write_runlog(ERROR, "StartOrStopInstanceByCommand failed:%s , errnum=%d, errno=%d.\n", cmd, ret, errno);
}
write_runlog(LOG, "StartOrStopInstanceByCommand success: %s.\n", cmd);
}
void StartOrStopNodeInstances(OperateType operateType)
{
if (g_currentNode->coordinate == 1) {
StartOrStopInstanceByCommand(operateType, g_currentNode->node, g_currentNode->DataPath);
}
for (uint32 ii = 0; ii < g_currentNode->datanodeCount; ii++) {
StartOrStopInstanceByCommand(
operateType, g_currentNode->node, g_currentNode->datanode[ii].datanodeLocalDataPath);
}
if (g_currentNode->gtm == 1) {
StartOrStopInstanceByCommand(operateType, g_currentNode->node, g_currentNode->gtmLocalDataPath);
}
}
* This is a node-level judgment whether the connection is ok or disconnected.
* CmaDisconnectWithAllRemoteCms : Judging the connection of cm_agent with other node cm_server is ok or disconnected.
* If cm_agent connect cm_server regadless of cms standy or primary, the node will not commit suicide.
* So, return true and maintain the agent_kill_instance_timeout.
* Conversely, return false, and the node will commit suicide and set the agent_kill_instance_timeout to 30 seconds.
*/
static bool CmaDisconnectWithAllRemoteCms()
{
write_runlog(DEBUG1, "cm_agent connect cm_server in other node: cmaConnectCmsInOtherNodeCount=%u.\n",
g_cmaConnectCmsInOtherNodeCount);
return (g_cmaConnectCmsInOtherNodeCount == 0) ? true : false;
}
* This is a AZ-level judgment whether the connection is ok or disconnected.
* CmaDisconnectWithAllRemoteCmsInOtherAz:
* Judging the connection of cm_agent with other AZ cm_server is ok or disconnected.
*/
static bool CmaDisconnectWithAllRemoteCmsInOtherAz()
{
bool isSingleAz = true;
for (uint32 i = 0; i < g_node_num; i++) {
if (strcmp(g_node[i].azName, g_currentNode->azName) != 0) {
isSingleAz = false;
break;
}
}
if (isSingleAz) {
write_runlog(DEBUG1, "CmaDisconnectWithAllRemoteCmsInOtherAz, isSingleAz=%d.\n", (int32)isSingleAz);
return false;
}
write_runlog(DEBUG1, "cm_agent connect cm_server in other AZ, cmaConnectCmsInOtherAzCount=%u.\n",
g_cmaConnectCmsInOtherAzCount);
return (g_cmaConnectCmsInOtherAzCount == 0) ? true : false;
}
bool isUpgradeCluster()
{
int rcs;
struct stat stat_buf;
char pg_host_path[MAX_PATH_LEN] = {0};
char grayscale_upgrade_check[MAX_PATH_LEN] = {0};
char upgrade_flag[MAX_PATH_LEN] = {0};
rcs = cmagent_getenv("PGHOST", pg_host_path, sizeof(pg_host_path));
if (rcs != EOK) {
write_runlog(ERROR, "get PGHOST failed!\n");
return false;
}
check_input_for_security(pg_host_path);
rcs = snprintf_s(grayscale_upgrade_check, MAX_PATH_LEN, MAX_PATH_LEN - 1, "%s/binary_upgrade", pg_host_path);
securec_check_intval(rcs, (void)rcs);
if (access(grayscale_upgrade_check, F_OK) == 0) {
rcs = snprintf_s(upgrade_flag, MAX_PATH_LEN, MAX_PATH_LEN - 1,
"%s/inplace_upgrade_flag", grayscale_upgrade_check);
securec_check_intval(rcs, (void)rcs);
if (stat(upgrade_flag, &stat_buf) == 0) {
return true;
}
rcs = snprintf_s(upgrade_flag, MAX_PATH_LEN, MAX_PATH_LEN - 1,
"%s/upgrade_step.csv", grayscale_upgrade_check);
securec_check_intval(rcs, (void)rcs);
if (stat(upgrade_flag, &stat_buf) == 0) {
return true;
}
}
return false;
}
* @brief get cluster upgrade mode.
* @return int NOT_UPGRADED/GRAYSCALE_UPGRADED/INPLACE_UPGRADED
*/
static int GetUpgradeMode()
{
int rcs;
struct stat statBufInPlace = {0};
struct stat statBufGary = {0};
char pgHostPath[MAX_PATH_LEN] = {0};
char upgradeFolder[MAX_PATH_LEN] = {0};
char inplaceUpgradeFlag[MAX_PATH_LEN] = {0};
char grayScaleUpgradeFlag[MAX_PATH_LEN] = {0};
rcs = cmagent_getenv("PGHOST", pgHostPath, sizeof(pgHostPath));
if (rcs != EOK) {
write_runlog(ERROR, "get PGHOST failed!\n");
return NOT_UPGRADED;
}
check_input_for_security(pgHostPath);
rcs = snprintf_s(upgradeFolder, MAX_PATH_LEN, MAX_PATH_LEN - 1, "%s/binary_upgrade", pgHostPath);
securec_check_intval(rcs, (void)rcs);
if (access(upgradeFolder, F_OK) != 0) {
write_runlog(LOG, "binary_upgrade: %s is not exist!\n", pgHostPath);
return NOT_UPGRADED;
}
rcs = snprintf_s(inplaceUpgradeFlag, MAX_PATH_LEN, MAX_PATH_LEN - 1, "%s/inplace_upgrade_flag", upgradeFolder);
securec_check_intval(rcs, (void)rcs);
if (stat(inplaceUpgradeFlag, &statBufInPlace) == 0) {
return INPLACE_UPGRADED;
}
rcs = snprintf_s(grayScaleUpgradeFlag, MAX_PATH_LEN, MAX_PATH_LEN - 1, "%s/upgrade_step.csv", upgradeFolder);
securec_check_intval(rcs, (void)rcs);
if (stat(grayScaleUpgradeFlag, &statBufGary) == 0) {
return GRAYSCALE_UPGRADED;
}
return NOT_UPGRADED;
}
* Check whether the instance in maintenance mode exists.
*
* The maintenance instance flag file path.
* - true: The maintenance instance flag file exists and the file contains content.
* - false: The maintenance instance flag file does not exist or the file is empty.
*/
bool is_maintenance_instance(const char *file_path)
{
char current[INSTANCE_ID_LEN] = { 0 };
bool instance_maintenance = false;
FILE *fd = fopen(file_path, "re");
if (!fd) {
write_runlog(DEBUG1, "Can't open the maintenance instance flag file: file_path=\"%s\","
"errno=\"[%d]\".\n", file_path, errno);
return instance_maintenance;
}
while (!feof(fd)) {
if (fscanf_s(fd, "%s\n", current, INSTANCE_ID_LEN) < 0) {
write_runlog(LOG, "Failed to get maintenance instance flag file content.\n");
break;
}
if (strlen(current) != 0) {
write_runlog(LOG, "Get maintenance instance flag file successfully: file_path=\"%s\","
" instance_id=\"%s\".\n", file_path, current);
instance_maintenance = true;
break;
}
}
(void)fclose(fd);
return instance_maintenance;
}
maintenance_mode getMaintenanceMode()
{
maintenance_mode mode = MAINTENANCE_MODE_NONE;
if (isUpgradeCluster()) {
mode = MAINTENANCE_MODE_UPGRADE;
} else if (is_maintenance_instance(instance_maintance_path)) {
mode = MAINTENANCE_MODE_DILATATION;
}
return mode;
}
static inline bool isDisableKillSelfInstances(const maintenance_mode &mode)
{
return mode == MAINTENANCE_MODE_UPGRADE;
}
bool isMaintenanceModeDisableOperation(const cma_operation op)
{
maintenance_mode mode = getMaintenanceMode();
bool isDisable = false;
if (op == CMA_KILL_SELF_INSTANCES) {
return isDisableKillSelfInstances(mode);
}
return isDisable;
}
bool isDisconnectTimeout(const struct timespec last, int timeout)
{
struct timespec now = {0, 0};
(void)clock_gettime(CLOCK_MONOTONIC, &now);
if (timeout == 0) {
return false;
} else {
return (now.tv_sec - last.tv_sec) >= timeout;
}
}
* @brief execute remote cmd function
* @param remoteNodeid remoteNodeid in cluster
* @param cmd command
*/
void ExecSsh(uint32 remoteNodeid, const char *cmd)
{
int rc;
char command[MAXPGPATH] = {0};
uint32 ii;
int ret;
for (ii = 0; ii < g_node[remoteNodeid].sshCount; ii++) {
ret = snprintf_s(command, MAXPGPATH, MAXPGPATH - 1, "pssh %s -s -H %s '%s'", PSSH_TIMEOUT_OPTION,
g_node[remoteNodeid].sshChannel[ii], cmd);
securec_check_intval(ret, (void)ret);
rc = system(command);
if (rc == 0) {
break;
} else {
write_runlog(ERROR,
"failed to execute the ssh command: nodeId=%u, command=\"%s\", systemReturn=%d,"
" commandReturn=%d, errno=%d\n",
g_node[remoteNodeid].node, command, rc, SHELL_RETURN_CODE(rc), errno);
return;
}
}
}
* @brief check version between agent and cmserver primary, if version is not equal kill all instance
*/
static void StopNodeSelfByVersionNum()
{
int rc;
char upgradeVersionFile[MAX_PATH_LEN] = {0};
char versonResultPath[MAX_PATH_LEN] = {0};
char localVersion[MAX_PATH_LEN] = {0};
char tmp[MAXPGPATH] = {0};
char command[MAXPGPATH] = {0};
char exec_path[MAX_PATH_LEN] = {0};
const int versionLength = 6;
if (GetHomePath(exec_path, sizeof(exec_path)) != 0) {
return;
}
rc = snprintf_s(upgradeVersionFile, MAX_PATH_LEN, MAX_PATH_LEN - 1, "%s/bin/%s", exec_path, "upgrade_version");
securec_check_intval(rc, (void)rc);
canonicalize_path(upgradeVersionFile);
rc = snprintf_s(versonResultPath, MAX_PATH_LEN, MAX_PATH_LEN - 1, "%s/bin/%s-%u", exec_path, CHECK_VERSION_RESULT,
g_currentNode->node);
securec_check_intval(rc, (void)rc);
FILE *fp = fopen(upgradeVersionFile, "re");
if (fp == NULL) {
write_runlog(ERROR, "failed to open File:%s\n", upgradeVersionFile);
return;
}
(void)fgets(tmp, MAXPGPATH, fp);
rc = memset_s(localVersion, MAX_PATH_LEN, 0, MAX_PATH_LEN);
securec_check_errno(rc, (void)rc);
(void)fgets(tmp, MAXPGPATH, fp);
rc = strncpy_s(localVersion, MAX_PATH_LEN, tmp, versionLength);
securec_check_errno(rc, (void)rc);
check_input_for_security(localVersion);
if (g_serverNodeId != g_currentNode->node) {
rc = snprintf_s(command, MAXPGPATH, MAXPGPATH - 1,
"args=$(sed -n '2p' %s); if [ $args = \"%s\" ]; then echo \"0\"; else touch %s; fi",
upgradeVersionFile, localVersion, versonResultPath);
securec_check_intval(rc, (void)rc);
ExecSsh(g_serverNodeId - 1, command);
write_runlog(LOG, "try to touch file:%s, localVersion: %s, command: %s\n",
CHECK_VERSION_RESULT, localVersion, command);
}
(void)fclose(fp);
}
uint64 GetTimeMinus(const struct timeval checkEnd, const struct timeval checkBegin)
{
const int secTomicSec = 1000000;
return (uint64)((checkEnd.tv_sec - checkBegin.tv_sec) * secTomicSec + (checkEnd.tv_usec - checkBegin.tv_usec));
}
void* ConnCmsPMain(void* arg)
{
(void)clock_gettime(CLOCK_MONOTONIC, &g_serverHeartbeatTime);
bool have_killed_nodes = false;
bool isToStopInstances = false;
struct timeval checkBeginFunction = {0, 0};
struct timeval checkEndFunction = {0, 0};
const int twoSec = 2;
for (;;) {
if (g_exitFlag || g_shutdownRequest) {
write_runlog(LOG, "[ConnCmsPMain] ConnCmsPMain shutdown request.\n");
break;
}
(void)gettimeofday(&checkBeginFunction, NULL);
isToStopInstances = false;
if (agent_cm_server_connect == NULL) {
write_runlog(LOG, "%d cm_agent connect to cm_server start.\n", __LINE__);
agent_cm_server_connect = GetConnToCmserver(0);
if (agent_cm_server_connect != NULL) {
g_firstConnectFlag = true;
g_cmServerNeedReconnect = false;
(void)clock_gettime(CLOCK_MONOTONIC, &g_serverHeartbeatTime);
have_killed_nodes = false;
g_agentConnCmsSuccess = true;
} else {
* If the connection is disconnected, we need to execute the operation of stopping instances.
* If the connection is ok, we still need to judge the condition where cma connect cms
* in other AZ is ok or disconnected. And when the connection is disconnected, we also
* execute the operation of stopping instances.
* CmaDisconnectWithAllRemoteCms:
* The cma in current node disconnect with all remote cms in other nodes.
* CmaDisconnectWithAllRemoteCmsInOtherAz:
* The cma in current AZ disconnect with all remote cms in other AZ.
*/
if (CmaDisconnectWithAllRemoteCms()) {
isToStopInstances = true;
} else if (CmaDisconnectWithAllRemoteCmsInOtherAz()) {
isToStopInstances = true;
}
* If isToStopInstances is true, and g_enableFenceDn is true,
* agentStopInstanceDelayTime is FENCE_TIMEOUT, 30 seconds.
* If isToStopInstances is true, and g_enableFenceDn is false,
* agentStopInstanceDelayTime is DISABLE_TIMEOUT, 0 seconds, never timeout.
* If isToStopInstances is false, agentStopInstanceDelayTime is agent_kill_instance_timeout,
* 0 second by default,
* and the operation of stopping instances will not be executed.
*/
#ifndef ENABLE_MULTIPLE_NODES
uint32 timeout = IsBoolCmParamTrue(g_enableFenceDn) ? FENCE_TIMEOUT : DISABLE_TIMEOUT;
uint32 agentStopInstanceDelayTime = isToStopInstances ? timeout : agent_kill_instance_timeout;
#else
uint32 agentStopInstanceDelayTime = isToStopInstances ? DISABLE_TIMEOUT : agent_kill_instance_timeout;
#endif
if (isDisconnectTimeout(g_disconnectTime, (int)agentStopInstanceDelayTime) && !have_killed_nodes) {
if ((undocumentedVersion == 0) && isMaintenanceModeDisableOperation(CMA_KILL_SELF_INSTANCES)) {
have_killed_nodes = false;
write_runlog(LOG, "%d Maintaining cluster: cm agent cannot stop self instances.\n", __LINE__);
} else if (!g_firstConnectFlag) {
have_killed_nodes = false;
write_runlog(LOG, "Agent has never successfully connected to the server,"
" so can not stop instances of current node.\n");
} else {
write_runlog(LOG, "agent disconnect from cm_server %u seconds, stop instances in this node. "
"sync_dropped_coordinator change to false.\n", agentStopInstanceDelayTime);
if (g_isPauseArbitration) {
continue;
}
g_syncDroppedCoordinator = false;
have_killed_nodes = true;
#ifndef ENABLE_MULTIPLE_NODES
* Kill datanode proccess, so that it can be restarted with pending mode.
*/
uint32 i;
for (i = 0; i < g_currentNode->datanodeCount; i++) {
immediate_stop_one_instance(g_currentNode->datanode[i].datanodeLocalDataPath, INSTANCE_DN);
}
#endif
}
}
}
}
(void)gettimeofday(&checkEndFunction, NULL);
if ((checkEndFunction.tv_sec - checkBeginFunction.tv_sec) > twoSec) {
write_runlog(LOG, "[ConnCmsPMain] agent connect to server takes %llu.\n",
(unsigned long long)GetTimeMinus(checkEndFunction, checkBeginFunction));
}
CmUsleep(AGENT_RECV_CYCLE);
}
write_runlog(LOG, "[ConnCmsPMain] ConnCmsPMain exit.\n");
return NULL;
}
void* CheckUpgradeMode(void* arg)
{
for (;;) {
if (g_agentConnCmsSuccess && GetUpgradeMode() == NOT_UPGRADED) {
StopNodeSelfByVersionNum();
g_agentConnCmsSuccess = false;
}
CmUsleep(AGENT_RECV_CYCLE);
}
return NULL;
}