* 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_status_check.cpp
* cma process cms messages functions
*
* IDENTIFICATION
* src/cm_agent/cma_status_check.cpp
*
* -------------------------------------------------------------------------
*/
#include <mntent.h>
#include <errno.h>
#include <strings.h>
#include <sys/stat.h>
#include <sys/vfs.h>
#include <arpa/inet.h>
#include <netinet/in.h>
#include "cma_connect.h"
#include "cma_global_params.h"
#include "cma_common.h"
#include "cm_text.h"
#include "cma_client.h"
#include "cma_instance_management.h"
#include "cma_instance_management_res.h"
#include "cma_process_messages.h"
#include "cma_connect.h"
#include "cma_instance_check.h"
#include "cm_util.h"
#ifdef ENABLE_MULTIPLE_NODES
#include "cma_coordinator.h"
#endif
#ifdef ENABLE_XALARMD
#ifdef __cplusplus
extern "C" {
#endif
#include <xalarm/register_xalarm.h>
#ifdef __cplusplus
}
#endif
#include "cjson/cJSON.h"
#include "cma_xalarm_event_compat.h"
struct alarm_register *g_xalarmEventRegister = NULL;
static const uint32 XALARM_TYPE_OCCUR = 1;
static const uint32 XALARM_TYPE_RECOVER = 2;
static const uint32 XALARM_REBOOT_EVENT_ID = 1003;
static const uint32 XALARM_REBOOT_ACK_EVENT_ID = 1004;
static const uint32 XALARM_OOM_EVENT_ID = 1005;
static const uint32 XALARM_OOM_ACK_EVENT_ID = 1006;
static const uint32 XALARM_PANIC_EVENT_ID = 1007;
static const uint32 XALARM_PANIC_ACK_EVENT_ID = 1008;
static const uint32 XALARM_KERNEL_REBOOT_EVENT_ID = 1009;
static const uint32 XALARM_KERNEL_REBOOT_ACK_EVENT_ID = 1010;
static const uint32 XALARM_UBUS_MEM_EVENT_ID = 1013;
static const uint32 XALARM_EVENT_NODE_MAP_MAX = CM_NODE_MAXNUM;
static const uint32 XALARM_EVENT_MAP_TEXT_LEN = MAX_PATH_LEN * 4;
static const uint32 XALARM_EVENT_COUNT = 5;
static const uint32 INVALID_ALARM_NODE_ID = 0;
static const int DECIMAL_BASE = 10;
static const uint32 MICROSECONDS_PER_MILLISECOND = 1000;
static const uint32 CMS_PRIMARY_SWITCH_WAIT_TIMEOUT_MS = 30000;
typedef struct XalarmNodeMapItemT {
uint32 nodeId;
uint32 cna;
} XalarmNodeMapItem;
static XalarmNodeMapItem g_xalarmNodeMap[XALARM_EVENT_NODE_MAP_MAX];
static uint32 g_xalarmNodeMapCount = 0;
#endif
#include "cma_status_check.h"
#include "cm_ip.h"
#include "cm_msg_version_convert.h"
* dilatation status. If the cluster in dilatation status, we query coordinate and report status for every loop. Or,
* only query once. If you want to use this flag, you need to restart cm_agent. The default value is true means, we need
* to query coordinate at least once at the start time.
*/
const int ETCD_CHECK_TIMES = 3;
static const int THRESHOLD_FORMAT = 4;
static const int THRESHOLD_MAX_VALUE = 100;
static const int THRESHOLD_MIN_VALUE = 0;
static const int INFO_POS = 5;
static const uint64 SLOW_DISK_CHECK_PERIOD = 5 * 60;
static const uint64 SLOW_DISK_CHECK_THRESHOLD = SLOW_DISK_CHECK_PERIOD * 2 / 10;
static uint32 g_sys_report_interval = 1;
MemCheckInfo g_memoryCheckInfoList[] = {
{"MemTotal", "MemTotal: %lu kB", MEM_STAT_TOTAL},
{"MemFree", "MemFree: %lu kB", MEM_STAT_FREE},
{"MemAvailable", "MemAvailable: %lu kB", MEM_STAT_AVAILABLE},
{"Buffers", "Buffers: %lu kB", MEM_STAT_BUFFERS},
{"Cached", "Cached: %lu kB", MEM_STAT_CACHED}
};
SlowIotLevelInfo g_slowIoLevelInfoList[] = {
{SVCTM_LEVEL_SLIGHT, 0, 0},
{SVCTM_LEVEL_MODERATE, 10, 1},
{SVCTM_LEVEL_SERIOUS, 20, 2}
};
void etcd_status_check_and_report(void)
{
if (g_currentNode->etcd == 0) {
return;
}
cm_query_instance_status reportMsg = {0};
(void)pthread_rwlock_wrlock(&(g_etcdReportMsg.lk_lock));
errno_t rc = memcpy_s(&reportMsg, sizeof(cm_query_instance_status),
&(g_etcdReportMsg.report_msg), sizeof(cm_query_instance_status));
securec_check_errno(rc, (void)rc);
(void)pthread_rwlock_unlock(&(g_etcdReportMsg.lk_lock));
PushMsgToCmsSendQue((char *)&reportMsg, (uint32)sizeof(cm_query_instance_status), "etcd status");
}
static status_t GetCpuInfo(IoStat *stat, CpuInfo &cpu, const char *line)
{
int ret;
ret = sscanf_s(line + INFO_POS, "%lu %lu %lu %lu %lu %lu %lu %lu %lu %lu", &cpu.cpuUser, &cpu.cpuNice, &cpu.cpuSys,
&cpu.cpuIdle, &cpu.cpuIwait, &cpu.cpuHardirq, &cpu.cpuSoftirq, &cpu.cpuSteal, &cpu.cpuGuest, &cpu.cpuGuestNice);
if (ret == -1) {
write_runlog(ERROR, "get cpu info fail, result is %d.\n", ret);
return CM_ERROR;
}
if (stat != NULL) {
stat->idle = cpu.cpuIdle;
stat->uptime = cpu.cpuUser + cpu.cpuNice + cpu.cpuSys + cpu.cpuIdle + cpu.cpuIwait + cpu.cpuHardirq +
cpu.cpuSteal + cpu.cpuSoftirq;
}
return CM_SUCCESS;
}
static int ReadCpuStatus(int cpu_nr, IoStat *stat, bool getTotalCpuHave)
{
FILE* cpufp;
char line[8192];
CpuInfo cpu = {0};
if ((cpufp = fopen(FILE_CPUSTAT, "re")) == NULL) {
write_runlog(ERROR, "cannot open file: %s \n", FILE_CPUSTAT);
return -1;
}
while (fgets(line, sizeof(line), cpufp) != NULL) {
if (!strncmp(line, "cpu ", 4) && (cpu_nr == 1 || getTotalCpuHave)) {
if (GetCpuInfo(stat, cpu, line) != CM_SUCCESS) {
(void)fclose(cpufp);
return -1;
}
(void)fclose(cpufp);
uint64 tmp = cpu.cpuUser + cpu.cpuNice + cpu.cpuSys;
uint64 total = tmp + cpu.cpuIdle;
if (total == 0) {
write_runlog(ERROR, "abnormal cpu info.\n");
return -1;
}
return (int)((PERCENT * tmp) / total);
}
if (cpu_nr > 1 && !strncmp(line, "cpu0", 4)) {
if (GetCpuInfo(stat, cpu, line) != CM_SUCCESS) {
(void)fclose(cpufp);
return -1;
}
(void)fclose(cpufp);
return 0;
}
}
write_runlog(ERROR, "get cpu info fail.\n");
(void)fclose(cpufp);
return 0;
}
void ReadDiskstatsStatus(const char* device, IoStat* stat)
{
FILE* iofp;
char line[MAX_PATH_LEN] = {0};
char dev_name[MAX_DEVICE_DIR] = {0};
int i;
uint64 rd_ios, rd_merges_or_rd_sec, rd_ticks_or_wr_sec;
uint64 wr_ios, wr_merges, rd_sec_or_wr_ios, wr_sec;
uint32 major, minor;
uint32 rq_ticks, ios_pgr, wr_ticks;
if ((iofp = fopen(FILE_DISKSTAT, "re")) == NULL) {
write_runlog(ERROR, "failed to open file %s", FILE_DISKSTAT);
return;
}
while (fgets(line, MAX_PATH_LEN, iofp) != NULL) {
i = sscanf_s(line,
"%u %u %s %lu %lu %lu %lu %lu %lu %lu %u %u %lu %u",
&major,
&minor,
dev_name,
MAX_DEVICE_DIR - 1,
&rd_ios,
&rd_merges_or_rd_sec,
&rd_sec_or_wr_ios,
&rd_ticks_or_wr_sec,
&wr_ios,
&wr_merges,
&wr_sec,
&wr_ticks,
&ios_pgr,
&stat->tot_ticks,
&rq_ticks);
check_sscanf_s_result(i, 14);
securec_check_intval(i, (void)i);
if (i == 14) {
if (strcmp(dev_name, device) != 0) {
continue;
} else {
break;
}
}
}
if (*dev_name == '\0') {
write_runlog(LOG, "cannot get the information of the file %s.\n", FILE_DISKSTAT);
}
(void)fclose(iofp);
}
uint64 GetAverageValue(uint64 value1, uint64 value2, uint64 itv, uint32 unit)
{
if (itv == 0) {
return 0;
}
if ((value2 < value1) && (value1 <= 0xffffffff)) {
return (((value2 - value1) & 0xffffffff)) * unit / itv;
} else {
return ((value2 - value1) * unit / itv);
}
}
static uint64 ReadDiskIOStat(const char* device, int cpu_nr, IoStat* oldIoStatus, bool needWriteLog)
{
long ticks;
IoStat ioStatus = {0};
if ((ticks = sysconf(_SC_CLK_TCK)) == -1) {
write_runlog(ERROR, "get ticks fail.\n");
return 0;
}
uint32 hz = (unsigned int)ticks;
(void)ReadCpuStatus(cpu_nr, &ioStatus, false);
if (ioStatus.uptime == 0) {
write_runlog(LOG, "get cpu time iz 0.\n");
return 0;
}
uint64 totalTime = ioStatus.uptime - oldIoStatus->uptime;
uint64 idleTime = ioStatus.idle - oldIoStatus->idle;
if (oldIoStatus->uptime == 0 || totalTime == 0) {
write_runlog(DEBUG1, "uptime is %lu, old_uptime is %lu,\n", ioStatus.uptime, oldIoStatus->uptime);
oldIoStatus->uptime = ioStatus.uptime;
oldIoStatus->idle = ioStatus.idle;
ReadDiskstatsStatus(device, oldIoStatus);
return 0;
}
ReadDiskstatsStatus(device, &ioStatus);
const uint32 percent = 100;
uint64 ioUtil = GetAverageValue(oldIoStatus->tot_ticks, ioStatus.tot_ticks, totalTime, hz) / 10;
uint64 cpuUtil = percent * (totalTime - idleTime) / totalTime;
oldIoStatus->tot_ticks = ioStatus.tot_ticks;
oldIoStatus->uptime = ioStatus.uptime;
oldIoStatus->idle = ioStatus.idle;
if (ioUtil > PERCENT) {
ioUtil = PERCENT;
}
if (!needWriteLog) {
write_runlog(DEBUG1, "device %s, [Io util: %lu%%]\n", device, ioUtil);
return ioUtil;
}
if (ioUtil > 60) {
write_runlog(LOG, "device %s, [Cpu util: %lu%%], [Io util: %lu%%]\n", device, cpuUtil, ioUtil);
} else {
write_runlog(DEBUG1, "device %s, [Cpu util: %lu%%], [Io util: %lu%%]\n", device, cpuUtil, ioUtil);
}
return ioUtil;
}
static void CmGetDisk(const char* datadir, char* devicename, uint32 nameLen)
{
char dfcommand[MAX_PATH_LEN] = {0};
char devicePath[MAX_PATH_LEN] = {0};
errno_t rc = snprintf_s(dfcommand, MAX_PATH_LEN, MAX_PATH_LEN - 1, "df -h %s", datadir);
securec_check_intval(rc, (void)rc);
const char* mode = "r";
FILE* fp = popen(dfcommand, mode);
if (fp == NULL) {
write_runlog(ERROR, "execute %s fail\n", dfcommand);
return;
}
char buf[CM_MAX_COMMAND_LONG_LEN] = {0};
if (fgets(buf, sizeof(buf), fp) == NULL) {
(void)pclose(fp);
write_runlog(ERROR, "get first line fail.\n");
return;
} else {
write_runlog(LOG, "first line is %s.\n", buf);
}
if (fgets(buf, sizeof(buf), fp) != NULL) {
write_runlog(LOG, "second line is %s.\n", buf);
uint32 length = (uint32)strlen(buf);
if (length == 0) {
(void)pclose(fp);
write_runlog(LOG, "execute %s, result is empty.\n", dfcommand);
return;
}
uint32 lengthDevice = 0;
for (uint32 i = 0; i < length; i++) {
if (lengthDevice >= MAX_PATH_LEN - 1) {
(void)pclose(fp);
write_runlog(LOG, "length is not enough for etcd data path device.\n");
return;
}
if ((buf[i] == ' ' || buf[i] == 10)) {
break;
}
devicePath[lengthDevice] = buf[i];
lengthDevice++;
}
} else {
(void)pclose(fp);
write_runlog(ERROR, "get second line fail.\n");
return;
}
(void)pclose(fp);
rc = snprintf_s(dfcommand, MAX_PATH_LEN, MAX_PATH_LEN - 1, "ls -l %s", devicePath);
securec_check_intval(rc, (void)rc);
fp = popen(dfcommand, mode);
if (fp == NULL) {
write_runlog(ERROR, "execute %s fail\n", dfcommand);
return;
} else {
write_runlog(LOG, "execute %s success.\n", dfcommand);
}
if (fgets(buf, sizeof(buf), fp) != NULL) {
uint32 length = (uint32)strlen(buf);
bool findDevice = false;
if (length == 0) {
(void)pclose(fp);
write_runlog(LOG, "execute %s, result is empty.\n", dfcommand);
return;
}
uint lengthDevice = 0;
for (uint32 i = 0; i < length; i++) {
if (lengthDevice >= MAX_DEVICE_DIR - 1) {
(void)pclose(fp);
write_runlog(LOG, "length is not enough for etcd data path device.\n");
return;
}
if (buf[i] != '>' && lengthDevice == 0 && !findDevice) {
continue;
}
if (buf[i] == '>') {
findDevice = true;
continue;
}
if (findDevice && (lengthDevice != 0 || buf[i] == '/')) {
if (buf[i] == '/') {
i++;
}
if ((buf[i] == ' ' || buf[i] == 10)) {
break;
}
if (i < length && lengthDevice < nameLen) {
devicename[lengthDevice] = buf[i];
lengthDevice++;
}
}
}
if (findDevice) {
write_runlog(LOG, "device name is %s.\n", devicename);
(void)pclose(fp);
return;
}
}
(void)pclose(fp);
size_t buf_len = 0;
struct mntent* ent;
struct mntent tempEnt = {};
FILE *mtfp = fopen(FILE_MOUNTS, "re");
if (mtfp == NULL) {
write_runlog(LOG, "cannot open file %s.\n", FILE_MOUNTS);
return;
}
char *mntentBuffer = (char *)malloc(4 * FILENAME_MAX);
if (mntentBuffer == NULL) {
write_runlog(ERROR,
"Failed to allocate memory: Out of memory. RequestSize=%d.\n", 4 * FILENAME_MAX);
(void)fclose(mtfp);
return;
}
while ((ent = getmntent_r(mtfp, &tempEnt, mntentBuffer, 4 * FILENAME_MAX)) != NULL) {
buf_len = strlen(ent->mnt_fsname);
* get the file system with type of ext* or xfs.
* find the best fit for the data directory
*/
size_t offset = strlen("/dev/");
if (buf_len >= offset && strncmp(devicePath, "/dev/", offset) != 0) {
offset = 0;
}
if (strncmp(ent->mnt_fsname, devicePath, buf_len) == 0 && strlen(datadir) >= buf_len &&
buf_len == strlen(devicePath)) {
rc = strncpy_s(devicename, MAX_DEVICE_DIR, ent->mnt_fsname + offset, strlen(ent->mnt_fsname + offset));
if (rc != 0) {
write_runlog(ERROR, "memcpy device name fail.\n");
(void)fclose(mtfp);
FREE_AND_RESET(mntentBuffer);
return;
} else {
break;
}
}
}
write_runlog(LOG, "devicename is %s.\n", devicename);
(void)fclose(mtfp);
FREE_AND_RESET(mntentBuffer);
}
void GetDiskNameByDataPath(const char* datadir, char* devicename, uint32 nameLen)
{
return CmGetDisk(datadir, devicename, nameLen);
}
static int GetCpuCount(void)
{
char pathbuf[4096] = {0};
int ret = 0;
int cpucnt = 0;
errno_t rc;
if (access("/sys/devices/system", F_OK) == 0) {
do {
rc = snprintf_s(pathbuf, sizeof(pathbuf), sizeof(pathbuf) - 1, "/sys/devices/system/cpu/cpu%d", cpucnt);
securec_check_intval(rc, (void)rc);
ret = access(pathbuf, F_OK);
if (ret == 0) {
cpucnt++;
}
} while (ret == 0);
} else if (access("/proc/cpuinfo", F_OK) == 0) {
FILE* fd;
if ((fd = fopen("/proc/cpuinfo", "re")) == NULL) {
return -1;
}
while (fgets(pathbuf, sizeof(pathbuf), fd) != NULL) {
if (strncmp("processor", pathbuf, strlen("processor")) == 0) {
cpucnt++;
}
}
(void)fclose(fd);
}
return cpucnt ? cpucnt : -1;
}
void etcd_disk_quota_check(const char *instanceName, const char *etcdData)
{
char check_cmd[CM_MAX_COMMAND_LONG_LEN] = {0};
const uint64 warningDiskQuota = 8160437862;
int needWarning = 0;
int rcs;
rcs = sprintf_s(check_cmd, sizeof(check_cmd),
"if [ `ls -l \"%s/member/snap/db\" | awk '{print $5}'` -ge %lu ]; then echo '1'; else echo '0'; fi;",
etcdData, warningDiskQuota);
securec_check_intval(rcs, (void)rcs);
FILE *fp = popen(check_cmd, "r");
if (fp == NULL) {
write_runlog(ERROR, "etcd_disk_quota_check fail: %s\n", check_cmd);
return;
}
rcs = fscanf_s(fp, "%d", &needWarning);
if (rcs > 0) {
static int appear_cnt = 0;
if (needWarning == 1) {
if (appear_cnt++ % 5 == 0) {
appear_cnt = 1;
write_runlog(LOG, "etcd db files takes too much disk space.\n");
}
report_ddb_fail_alarm(ALM_AT_Fault, instanceName, 2, DB_ETCD);
} else {
if (appear_cnt != 0) {
write_runlog(LOG, "etcd db files takes normal disk space.\n");
appear_cnt = 0;
}
report_ddb_fail_alarm(ALM_AT_Resume, instanceName, 2, DB_ETCD);
}
} else {
write_runlog(LOG, "Failed to get etcd db files's disk space.\n");
}
(void)pclose(fp);
}
static void GetDdbCfgApi(DrvApiInfo *drvApiInfo, ServerSocket *server, uint32 serverLen)
{
drvApiInfo->nodeNum = serverLen - 1;
drvApiInfo->serverList = server;
drvApiInfo->serverLen = serverLen;
drvApiInfo->modId = MOD_CMA;
drvApiInfo->nodeId = g_currentNode->node;
drvApiInfo->client_t.tlsPath = &g_tlsPath;
drvApiInfo->timeOut = DDB_DEFAULT_TIMEOUT;
}
static void SetServerSocketWithEtcdInfo(ServerSocket *server, staticNodeConfig *node)
{
server->nodeIdInfo.azName = node->azName;
server->nodeIdInfo.nodeId = node->node;
server->nodeIdInfo.instd = node->etcdId;
server->nodeInfo.nodeName = node->etcdName;
server->nodeInfo.len = CM_NODE_NAME;
server->host = node->etcdClientListenIPs[0];
server->port = node->etcdClientListenPort;
}
int CheckCertFilePermission(const char *certFile)
{
struct stat buf;
if (stat(certFile, &buf) != 0) {
write_runlog(WARNING, "Try to stat cert key file \"%s\" failed!\n", certFile);
return -1;
}
if (!S_ISREG(buf.st_mode) || (buf.st_mode & (S_IRWXG | S_IRWXO)) || ((buf.st_mode & S_IRWXU) == S_IRWXU)) {
write_runlog(WARNING, "The file \"%s\" permission should be u=rw(600) or less.\n", certFile);
return -1;
}
return 0;
}
bool EtcdCertFileExpire(const char *certFile)
{
int rcs;
const int expireDays = 90;
int expireSeconds = expireDays * 24 * 60 * 60;
char command[CM_PATH_LENGTH] = {0};
char result[CM_PATH_LENGTH] = {0};
const char *certExpire = "Certificate will expire";
rcs = snprintf_s(command, CM_PATH_LENGTH, CM_PATH_LENGTH - 1, "openssl x509 -in %s -checkend %d 2>&1 &", certFile,
expireSeconds);
securec_check_intval(rcs, (void)rcs);
if (!ExecuteCmdWithResult(command, result, CM_PATH_LENGTH)) {
write_runlog(WARNING, "Execute check etcd cert file expire cmd %s failed, result=%s\n", command, result);
return false;
}
if (strstr(result, certExpire) != NULL) {
write_runlog(WARNING,
"etcd cert file %s may has been expired or will be expired in less %d days, please check!\n", certFile,
expireDays);
return true;
}
return false;
}
void CheckEtcdClientCertFile()
{
(void)CheckCertFilePermission(g_tlsPath.caFile);
(void)CheckCertFilePermission(g_tlsPath.crtFile);
(void)CheckCertFilePermission(g_tlsPath.keyFile);
(void)EtcdCertFileExpire(g_tlsPath.caFile);
}
void CheckEtcdServerCertFile()
{
int rcs;
char caFile[MAX_PATH_LEN] = {0};
char keyFile[MAX_PATH_LEN] = {0};
rcs = snprintf_s(caFile, MAX_PATH_LEN, MAX_PATH_LEN - 1, "%s/etcd.crt", g_currentNode->etcdDataPath);
securec_check_intval(rcs, (void)rcs);
rcs = snprintf_s(keyFile, MAX_PATH_LEN, MAX_PATH_LEN - 1, "%s/etcd.key", g_currentNode->etcdDataPath);
securec_check_intval(rcs, (void)rcs);
(void)CheckCertFilePermission(caFile);
(void)CheckCertFilePermission(keyFile);
(void)EtcdCertFileExpire(caFile);
}
void CheckEtcdCertFile()
{
static int checkTimes = 0;
const int everyCheckTime = 60;
if (checkTimes >= everyCheckTime) {
CheckEtcdClientCertFile();
CheckEtcdServerCertFile();
checkTimes = 0;
} else {
checkTimes++;
}
}
void* ETCDStatusCheckMain(void* arg)
{
thread_name = "ETCD_CHECK";
pthread_t threadId = pthread_self();
write_runlog(LOG, "etcd status check thread start, threadid %lu.\n", threadId);
int etcdReportFre = ETCD_NODE_UNHEALTH_FRE;
char devicename[MAX_DEVICE_DIR] = {0};
CmGetDisk(g_currentNode->etcdDataPath, devicename, MAX_DEVICE_DIR);
uint64 devicenamelength = strlen(devicename);
IoStat ioStatus = {0};
int cpu_nr = GetCpuCount();
errno_t rc;
cm_query_instance_status cm_query_instance_status_content = {0};
cm_query_instance_status_content.nodeId = g_currentNode->node;
cm_query_instance_status_content.msg_type = MSG_CM_QUERY_INSTANCE_STATUS;
cm_query_instance_status_content.msg_step = QUERY_STATUS_CMAGENT_STEP;
cm_query_instance_status_content.instanceType = PROCESS_ETCD;
cm_query_instance_status_content.pending = false;
char instanceName[CM_NODE_NAME] = {0};
rc = snprintf_s(
instanceName, sizeof(instanceName), sizeof(instanceName) - 1, "%s_%u", "etcd", g_currentNode->etcdId);
securec_check_intval(rc, (void)rc);
char command[MAXPGPATH * 2] = {0};
int checkInvalidEtcdTimes = 0;
const uint32 serverLen = 2;
ServerSocket server[serverLen] = {{0}};
SetServerSocketWithEtcdInfo(&server[0], g_currentNode);
server[1].host = NULL;
status_t st = CM_SUCCESS;
DdbInitConfig config = {DB_ETCD};
GetDdbCfgApi(&config.drvApiInfo, server, serverLen);
DdbNodeState nodeState;
for (;;) {
if (devicenamelength > 0) {
(void)ReadDiskIOStat(devicename, cpu_nr, &ioStatus, true);
}
CheckEtcdCertFile();
if (g_shutdownRequest || (agent_cm_server_connect == NULL) || g_exitFlag) {
write_runlog(LOG, "receive exit request in cma ETCDStatusCheckMain.\n");
cm_sleep(5);
continue;
}
if (cpu_nr <= 0) {
cpu_nr = GetCpuCount();
}
int tryTime = 0;
int tryTime1 = 0;
rc = memset_s(&nodeState, sizeof(DdbNodeState), 0, sizeof(DdbNodeState));
securec_check_errno(rc, (void)rc);
DdbConn dbCon = {0};
st = InitDdbConn(&dbCon, &config);
if (st != CM_SUCCESS) {
(void)pthread_rwlock_wrlock(&(g_etcdReportMsg.lk_lock));
cm_query_instance_status_content.status = CM_ETCD_DOWN;
rc = memcpy_s((void*)&(g_etcdReportMsg.report_msg),
sizeof(cm_query_instance_status_content),
(void*)&cm_query_instance_status_content,
sizeof(cm_query_instance_status_content));
securec_check_errno(rc, (void)rc);
(void)pthread_rwlock_unlock(&(g_etcdReportMsg.lk_lock));
write_runlog(ERROR, "etcd open failed when query etcd status. %s\n", DdbGetLastError(&dbCon));
cm_sleep(agent_check_interval);
continue;
}
do {
st = DdbInstanceState(&dbCon, g_currentNode->etcdName, &nodeState);
if (st != CM_SUCCESS) {
write_runlog(FATAL, "get ddb instance state failed, error is %s\n", DdbGetLastError(&dbCon));
}
tryTime1++;
if (st != CM_SUCCESS) {
if (nodeState.health == DDB_STATE_HEALTH) {
do {
st = DdbInstanceState(&dbCon, g_currentNode->etcdName, &nodeState);
tryTime++;
if (st != CM_SUCCESS) {
write_runlog(FATAL, "ddb instance is health, get state failed, error is %s\n",
DdbGetLastError(&dbCon));
}
} while ((st != CM_SUCCESS) && tryTime <= ETCD_CHECK_TIMES);
}
}
} while ((st != CM_SUCCESS) && tryTime1 <= ETCD_CHECK_TIMES);
if ((st != CM_SUCCESS) || (nodeState.health != DDB_STATE_HEALTH && nodeState.role == DDB_ROLE_LEADER)) {
checkInvalidEtcdTimes++;
} else {
checkInvalidEtcdTimes = 0;
}
if (checkInvalidEtcdTimes >= CHECK_INVALID_ETCD_TIMES) {
write_runlog(
ERROR, "can't get majority etcd state, but local is unhealthy and leader, will kill local etcd now.\n");
checkInvalidEtcdTimes = 0;
rc = strcpy_s(command, 2 * MAXPGPATH, SYSTEMQUOTE "killall etcd > /dev/null 2>&1 &" SYSTEMQUOTE);
securec_check_errno(rc, (void)rc);
int rct = system(command);
if (rct != -1) {
write_runlog(LOG, "killall etcd result is %d, shell result is %d.\n", rc, WEXITSTATUS(rc));
} else {
char error_buffer[ERROR_LIMIT_LEN] = {0};
(void)strerror_r(errno, error_buffer, ERROR_LIMIT_LEN);
write_runlog(ERROR, "Failed to call the system function: error=\"[%d] %s\","
" function=\"%s\", command=\"%s\".\n", errno, error_buffer, "system", command);
}
}
if (DdbFreeConn(&dbCon) != CM_SUCCESS) {
write_runlog(WARNING, "etcd_close failed,%s\n", DdbGetLastError(&dbCon));
}
if (st != CM_SUCCESS) {
cm_query_instance_status_content.status = CM_ETCD_DOWN;
} else {
if (nodeState.role == DDB_ROLE_LEADER) {
write_runlog(DEBUG1, "etcd state is StateLeader.\n");
cm_query_instance_status_content.status = CM_ETCD_LEADER;
} else if (nodeState.role == DDB_ROLE_FOLLOWER) {
write_runlog(DEBUG1, "etcd state is StateFollower.\n");
cm_query_instance_status_content.status = CM_ETCD_FOLLOWER;
}
}
if (cm_query_instance_status_content.status == CM_ETCD_DOWN) {
if (etcdReportFre > ETCD_NODE_UNHEALTH_FRE) {
etcdReportFre = ETCD_NODE_UNHEALTH_FRE;
}
if (etcdReportFre > 0) {
etcdReportFre--;
}
if (etcdReportFre <= 0) {
report_ddb_fail_alarm(ALM_AT_Fault, instanceName, 1, DB_ETCD);
}
} else {
etcdReportFre = ETCD_NODE_UNHEALTH_FRE;
report_ddb_fail_alarm(ALM_AT_Resume, instanceName, 1, DB_ETCD);
}
(void)pthread_rwlock_wrlock(&(g_etcdReportMsg.lk_lock));
rc = memcpy_s((void*)&(g_etcdReportMsg.report_msg), sizeof(cm_query_instance_status_content),
(void*)&cm_query_instance_status_content,
sizeof(cm_query_instance_status_content));
securec_check_errno(rc, (void)rc);
(void)pthread_rwlock_unlock(&(g_etcdReportMsg.lk_lock));
etcd_disk_quota_check(instanceName, g_currentNode->etcdDataPath);
cm_sleep(10);
continue;
}
}
void kerberos_status_check_and_report()
{
if (agent_cm_server_connect == NULL) {
return;
}
char kerberosConfigPath[MAX_PATH_LEN] = {0};
if (cmagent_getenv("MPPDB_KRB5_FILE_PATH", kerberosConfigPath, sizeof(kerberosConfigPath)) != EOK) {
write_runlog(DEBUG1, "kerberos_status_check_and_report: MPPDB_KRB5_FILE_PATH get fail.\n");
return;
}
struct stat stat_buf = {0};
if (stat(kerberosConfigPath, &stat_buf) != 0) {
write_runlog(DEBUG1, "kerberos_status_check_and_report: kerberos config file not exist.\n");
return;
}
agent_to_cm_kerberos_status_report reportMsg = {0};
(void)pthread_rwlock_wrlock(&(g_kerberosReportMsg.lk_lock));
errno_t rc = memcpy_s(&reportMsg, sizeof(agent_to_cm_kerberos_status_report),
&(g_kerberosReportMsg.report_msg), sizeof(agent_to_cm_kerberos_status_report));
securec_check_errno(rc, (void)rc);
(void)pthread_rwlock_unlock(&(g_kerberosReportMsg.lk_lock));
PushMsgToCmsSendQue((char *)&reportMsg, (uint32)sizeof(agent_to_cm_kerberos_status_report), "kerberos status");
}
static void SendDnReportMsg(const DnStatus *pkgDnStatus, uint32 datanodeId)
{
if (undocumentedVersion != 0 && undocumentedVersion < SUPPORT_IPV6_VERSION) {
agent_to_cm_datanode_status_report_ipv4 reportMsg = {0};
AgentToCmDatanodeStatusReportV2ToV1(&pkgDnStatus->reportMsg, &reportMsg);
write_runlog(DEBUG5, "dn(%u) reportMsg will send to cms.\n", datanodeId);
PushMsgToCmsSendQue((char *)&reportMsg, (uint32)sizeof(agent_to_cm_datanode_status_report_ipv4), "dn report");
} else {
agent_to_cm_datanode_status_report reportMsg = {0};
errno_t rc = memcpy_s(&reportMsg, sizeof(agent_to_cm_datanode_status_report),
&pkgDnStatus->reportMsg, sizeof(agent_to_cm_datanode_status_report));
securec_check_errno(rc, (void)rc);
write_runlog(DEBUG5, "dn(%u) reportMsg will send to cms.\n", datanodeId);
PushMsgToCmsSendQue((char *)&reportMsg, (uint32)sizeof(agent_to_cm_datanode_status_report), "dn report");
}
}
static void SendBarrierMsg(const DnStatus *pkgDnStatus, uint32 datanodeId)
{
AgentToCmBarrierStatusReport barrierMsg = {0};
errno_t rc = memcpy_s(&barrierMsg, sizeof(AgentToCmBarrierStatusReport),
&pkgDnStatus->barrierMsg, sizeof(AgentToCmBarrierStatusReport));
securec_check_errno(rc, (void)rc);
write_runlog(DEBUG5, "dn(%u) barrier(%d) will send to cms.\n", datanodeId, (int)pkgDnStatus->barrierMsgType);
PushMsgToCmsSendQue((char *)&barrierMsg, (uint32)sizeof(AgentToCmBarrierStatusReport), "dn barrier");
}
static void SendLpInfoMsg(const DnStatus *pkgDnStatus, uint32 datanodeId)
{
AgentCmDnLocalPeer lpInfo = {0};
errno_t rc = memcpy_s(&lpInfo, sizeof(AgentCmDnLocalPeer), &pkgDnStatus->lpInfo, sizeof(AgentCmDnLocalPeer));
securec_check_errno(rc, (void)rc);
write_runlog(DEBUG5, "dn(%u) dnLocalPeer will send to cms.\n", datanodeId);
PushMsgToCmsSendQue((char *)&lpInfo, (uint32)sizeof(AgentCmDnLocalPeer), "dnLocalPeer");
}
static void SendDiskUsageMsg(const DnStatus *pkgDnStatus, uint32 datanodeId)
{
AgentToCmDiskUsageStatusReport diskUsageMsg = {0};
errno_t rc = memcpy_s(&diskUsageMsg, sizeof(AgentToCmDiskUsageStatusReport),
&pkgDnStatus->diskUsageMsg, sizeof(AgentToCmDiskUsageStatusReport));
securec_check_errno(rc, (void)rc);
write_runlog(DEBUG5, "dn(%u) dnDiskUsage will send to cms.\n", datanodeId);
PushMsgToCmsSendQue((char *)&diskUsageMsg, (uint32)sizeof(AgentToCmDiskUsageStatusReport), "dnDiskUsage");
}
static void SendFloatIpMsg(const CmaDnFloatIpInfo *floatIpInfo, uint32 dnId)
{
if (!IsNeedCheckFloatIp() || (agent_backup_open != CLUSTER_PRIMARY)) {
return;
}
if (floatIpInfo->info.count == 0) {
return;
}
write_runlog(DEBUG5, "dn(%u) floatIpMsg will send to cms.\n", dnId);
PushMsgToCmsSendQue((const char *)floatIpInfo, (uint32)sizeof(CmaDnFloatIpInfo), "dn floatIpMsg");
}
static void SendDnReportMsgCore(const DnStatus *pkgDnStatus, uint32 datanodeId, AgentToCmserverDnSyncList *syncListMsg)
{
SendDnReportMsg(pkgDnStatus, datanodeId);
SendFloatIpMsg(&(pkgDnStatus->floatIpInfo), datanodeId);
if (g_clusterType == V3SingleInstCluster) {
return;
}
#if ((defined(ENABLE_MULTIPLE_NODES)) || (defined(ENABLE_PRIVATEGAUSS)))
write_runlog(DEBUG5, "dn(%u) syncListMsg will send to cms.\n", datanodeId);
PushMsgToCmsSendQue((char *)syncListMsg, (uint32)sizeof(AgentToCmserverDnSyncList), "dn syncListMsg");
#endif
if (pkgDnStatus->barrierMsgType == MSG_AGENT_CM_DATANODE_INSTANCE_BARRIER) {
SendBarrierMsg(pkgDnStatus, datanodeId);
return;
}
SendDiskUsageMsg(pkgDnStatus, datanodeId);
if (pkgDnStatus->reportMsg.receive_status.local_role != INSTANCE_ROLE_CASCADE_STANDBY) {
return;
}
SendLpInfoMsg(pkgDnStatus, datanodeId);
}
static void DnStatusFinalProcessing(DnStatus* pkgDnStatus, uint32 dnId)
{
if (pkgDnStatus->reportMsg.local_status.db_state == INSTANCE_HA_STATE_NORMAL ||
pkgDnStatus->reportMsg.local_status.db_state == INSTANCE_HA_STATE_NEED_REPAIR ||
pkgDnStatus->reportMsg.local_status.db_state == INSTANCE_HA_STATE_UNKONWN ||
(pkgDnStatus->reportMsg.local_status.db_state == INSTANCE_HA_STATE_DEMOTING &&
!g_isStorageWithDMSorDSS)) {
g_dnRoleForPhonyDead[dnId] = pkgDnStatus->reportMsg.local_status.local_role;
} else {
g_dnRoleForPhonyDead[dnId] = INSTANCE_ROLE_INIT;
}
if (g_dnPhonyDeadD[dnId] || g_dnCore[dnId]) {
pkgDnStatus->reportMsg.local_status.local_role = INSTANCE_ROLE_UNKNOWN;
write_runlog(WARNING, "datenode phony dead D or Core, set local_role Unknown\n");
}
if (g_dnPingFault[dnId]) {
pkgDnStatus->reportMsg.local_status.local_role = INSTANCE_ROLE_UNKNOWN;
pkgDnStatus->reportMsg.local_status.db_state = INSTANCE_HA_STATE_UNKONWN;
write_runlog(WARNING, "datenode ping fault, set local_role Unknown\n");
}
}
static void CopyDnReportMsg(AgentToCmserverDnSyncList *syncList, uint32 ii)
{
(void)pthread_rwlock_wrlock(&(g_dnSyncListInfo[ii].lk_lock));
errno_t rcs = memcpy_s(syncList, sizeof(AgentToCmserverDnSyncList), &(g_dnSyncListInfo[ii].dnSyncListMsg),
sizeof(AgentToCmserverDnSyncList));
securec_check_errno(rcs, (void)rcs);
g_dnSyncListInfo[ii].dnSyncListMsg.syncDone = FAILED_SYNC_DATA;
(void)pthread_rwlock_unlock(&(g_dnSyncListInfo[ii].lk_lock));
}
void DatanodeStatusReport(void)
{
errno_t rc;
for (uint32 ii = 0; ii < g_currentNode->datanodeCount; ii++) {
if (agent_cm_server_connect == NULL) {
continue;
}
(void)pthread_rwlock_wrlock(&(g_dnReportMsg[ii].lk_lock));
if (g_dnReportMsg[ii].dnStatus.reportMsg.local_status.local_role == INSTANCE_ROLE_PENDING ||
(g_dnReportMsg[ii].dnStatus.reportMsg.processStatus != INSTANCE_PROCESS_RUNNING &&
g_dnReportMsg[ii].dnStatus.reportMsg.connectStatus != AGENT_TO_INSTANCE_CONNECTION_OK)) {
if (g_dnPhonyDeadTimes[ii] != 0) {
g_dnPhonyDeadTimes[ii] = 0;
write_runlog(
LOG, "reset dn(%u) phony dead time to zero.\n", g_dnReportMsg[ii].dnStatus.reportMsg.instanceId);
}
}
if (g_dnReportMsg[ii].dnStatus.reportMsg.connectStatus == AGENT_TO_INSTANCE_CONNECTION_OK) {
DnStatus dnStatus;
rc = memcpy_s(&(dnStatus), sizeof(DnStatus), &(g_dnReportMsg[ii].dnStatus), sizeof(DnStatus));
securec_check_errno(rc, (void)rc);
(void)pthread_rwlock_unlock(&(g_dnReportMsg[ii].lk_lock));
AgentToCmserverDnSyncList syncList = {0};
CopyDnReportMsg(&syncList, ii);
DnStatusFinalProcessing(&dnStatus, ii);
dnStatus.reportMsg.phony_dead_times = g_dnPhonyDeadTimes[ii];
SendDnReportMsgCore(&dnStatus, g_currentNode->datanode[ii].datanodeId, &syncList);
} else if (g_dnReportMsg[ii].dnStatus.reportMsg.processStatus == INSTANCE_PROCESS_RUNNING) {
(void)pthread_rwlock_unlock(&(g_dnReportMsg[ii].lk_lock));
g_dnRoleForPhonyDead[ii] = INSTANCE_ROLE_INIT;
agent_to_cm_heartbeat hbMsg = {0};
hbMsg.msg_type = (int)MSG_AGENT_CM_HEARTBEAT;
hbMsg.node = g_currentNode->node;
hbMsg.instanceId = g_currentNode->datanode[ii].datanodeId;
hbMsg.instanceType = INSTANCE_TYPE_DATANODE;
PushMsgToCmsSendQue((char *)&hbMsg, (uint32)sizeof(agent_to_cm_heartbeat), "dn heartbeat");
} else {
DnStatus pkgDnStatus;
rc = memcpy_s(&(pkgDnStatus), sizeof(DnStatus), &(g_dnReportMsg[ii].dnStatus), sizeof(DnStatus));
securec_check_errno(rc, (void)rc);
(void)pthread_rwlock_unlock(&(g_dnReportMsg[ii].lk_lock));
AgentToCmserverDnSyncList syncList = {0};
CopyDnReportMsg(&syncList, ii);
g_dnRoleForPhonyDead[ii] = pkgDnStatus.reportMsg.local_status.local_role;
SendDnReportMsgCore(&pkgDnStatus, g_currentNode->datanode[ii].datanodeId, &syncList);
}
}
}
void fenced_UDF_status_check_and_report(void)
{
if (agent_cm_server_connect == NULL) {
return;
}
agent_to_cm_fenced_UDF_status_report reportMsg = {0};
reportMsg.msg_type = (int)MSG_AGENT_CM_FENCED_UDF_INSTANCE_STATUS;
reportMsg.nodeid = g_nodeId;
if (!g_fencedUdfStopped) {
reportMsg.status = INSTANCE_ROLE_NORMAL;
} else {
reportMsg.status = INSTANCE_ROLE_UNKNOWN;
}
write_runlog(DEBUG5, "node(%u) fenced UDF status will send to cms.\n", reportMsg.nodeid);
PushMsgToCmsSendQue((char *)&reportMsg, (uint32)sizeof(agent_to_cm_fenced_UDF_status_report), "fenced UDF status");
}
void InitReportMsg(agent_to_cm_datanode_status_report *reportMsg, int index)
{
errno_t rc =
memset_s(reportMsg, sizeof(agent_to_cm_datanode_status_report), 0, sizeof(agent_to_cm_datanode_status_report));
securec_check_errno(rc, (void)rc);
reportMsg->msg_type = MSG_AGENT_CM_DATA_INSTANCE_REPORT_STATUS;
reportMsg->node = g_currentNode->node;
reportMsg->instanceId = g_currentNode->datanode[index].datanodeId;
reportMsg->instanceType = INSTANCE_TYPE_DATANODE;
reportMsg->dn_restart_counts = g_dnReportMsg[index].dnStatus.reportMsg.dn_restart_counts;
}
void InitDnLocalPeerMsg(AgentCmDnLocalPeer *lpInfo, int32 index)
{
errno_t rc = memset_s(lpInfo, sizeof(AgentCmDnLocalPeer), 0, sizeof(AgentCmDnLocalPeer));
securec_check_errno(rc, (void)rc);
lpInfo->msgType = (int32)MSG_AGENT_CM_DATANODE_LOCAL_PEER;
lpInfo->instanceId = g_currentNode->datanode[index].datanodeId;
lpInfo->node = g_currentNode->node;
lpInfo->instanceType = INSTANCE_TYPE_DATANODE;
}
static void SetDnBaseMsg(BaseInstInfo *baseInfo, int32 index, int32 msgType)
{
baseInfo->msgType = msgType;
baseInfo->instId = g_currentNode->datanode[index].datanodeId;
baseInfo->node = g_currentNode->node;
baseInfo->instType = INSTANCE_TYPE_DATANODE;
}
static void InitDnFloatIpMsg(CmaDnFloatIpInfo *ipInfo, int32 index)
{
errno_t rc = memset_s(ipInfo, sizeof(CmaDnFloatIpInfo), 0, sizeof(CmaDnFloatIpInfo));
securec_check_errno(rc, (void)rc);
SetDnBaseMsg(&(ipInfo->baseInfo), index, (int32)MSG_AGENT_CM_FLOAT_IP);
}
void InitDNStatus(DnStatus *dnStatus, int i)
{
InitReportMsg(&dnStatus->reportMsg, i);
InitDnLocalPeerMsg(&(dnStatus->lpInfo), i);
InitDnFloatIpMsg(&(dnStatus->floatIpInfo), i);
}
void InitWrFloatIp(CmaWrFloatIp* wrFloatIp, int32 index)
{
wrFloatIp->msgType = MSG_AGENT_CM_WR_FLOAT_IP;
wrFloatIp->node = g_nodeId;
wrFloatIp->instId = RES_INSTANCE_ID_MIN + g_currentNode ->node;
wrFloatIp->count = 0;
for (uint32 i = 0; i < MAX_FLOAT_IP_COUNT; i++) {
wrFloatIp->netState[i] = NETWORK_STATE_UNKNOWN;
}
}
void ReportCmsWrFloatIp(CmaWrFloatIp* wrFloatIp, int32 index)
{
wrFloatIp->node = g_nodeId;
PushMsgToCmsSendQue((char *)wrFloatIp, (uint32)sizeof(CmaWrFloatIp), "wr float ip");
}
static void ChangeLocalRoleInBackup(int dnIdx, int *localDnRole)
{
if (*localDnRole == INSTANCE_ROLE_PRIMARY) {
write_runlog(ERROR, "dn_%u is Primary in cluster standby.\n", g_currentNode->datanode[dnIdx].datanodeId);
immediate_stop_one_instance(g_currentNode->datanode[dnIdx].datanodeLocalDataPath, INSTANCE_DN);
}
if (*localDnRole == INSTANCE_ROLE_MAIN_STANDBY) {
*localDnRole = INSTANCE_ROLE_PRIMARY;
} else if (*localDnRole == INSTANCE_ROLE_CASCADE_STANDBY) {
*localDnRole = INSTANCE_ROLE_STANDBY;
}
}
static void SendAlarmMsg(int alarmIndex, const char* logicClusterName, const char *instanceName, AlarmType type)
{
AlarmAdditionalParam tempAdditionalParam;
if ((g_abnormalAlarmList != NULL) && (!g_suppressAlarm)) {
if (type == ALM_AT_Resume) {
WriteAlarmAdditionalInfo(&tempAdditionalParam,
instanceName, "", "", logicClusterName,
&(g_abnormalAlarmList[alarmIndex]), ALM_AT_Resume);
AlarmReporter(&(g_abnormalAlarmList[alarmIndex]), ALM_AT_Resume, &tempAdditionalParam);
} else {
WriteAlarmAdditionalInfo(&tempAdditionalParam,
instanceName, "", "", logicClusterName,
&(g_abnormalAlarmList[alarmIndex]), ALM_AT_Fault, instanceName);
AlarmReporter(&(g_abnormalAlarmList[alarmIndex]), ALM_AT_Fault, &tempAdditionalParam);
}
}
return;
}
static void CheckAlmLocalPrimary(
DnStatus *dnStatus, int alarmIndex, const char *logicClusterName, const char *instanceName)
{
bool peerRoleStandby = (dnStatus->reportMsg.sender_status[0].peer_role == INSTANCE_ROLE_STANDBY ||
(agent_backup_open == CLUSTER_STREAMING_STANDBY &&
dnStatus->reportMsg.sender_status[0].peer_role == INSTANCE_ROLE_CASCADE_STANDBY));
if (peerRoleStandby) {
if ((dnStatus->reportMsg.local_status.db_state == INSTANCE_HA_STATE_NORMAL) &&
((dnStatus->reportMsg.sender_status[0].peer_state == INSTANCE_HA_STATE_NORMAL) ||
(dnStatus->reportMsg.sender_status[0].peer_state == INSTANCE_HA_STATE_CATCH_UP))) {
SendAlarmMsg(alarmIndex, logicClusterName, instanceName, ALM_AT_Resume);
} else {
SendAlarmMsg(alarmIndex, logicClusterName, instanceName, ALM_AT_Fault);
}
} else {
if (dnStatus->reportMsg.sender_status[0].peer_role != INSTANCE_ROLE_PENDING) {
SendAlarmMsg(alarmIndex, logicClusterName, instanceName, ALM_AT_Fault);
}
}
return;
}
static void CheckAlmLocalStandby(
DnStatus *dnStatus, int alarmIndex, const char *logicClusterName, const char *instanceName)
{
bool peerRolePrimary = (dnStatus->reportMsg.receive_status.peer_role == INSTANCE_ROLE_PRIMARY ||
(agent_backup_open == CLUSTER_STREAMING_STANDBY &&
dnStatus->reportMsg.receive_status.peer_role == INSTANCE_ROLE_MAIN_STANDBY));
if (peerRolePrimary) {
if ((dnStatus->reportMsg.receive_status.peer_state == INSTANCE_HA_STATE_NORMAL) &&
((dnStatus->reportMsg.local_status.db_state == INSTANCE_HA_STATE_NORMAL) ||
(dnStatus->reportMsg.local_status.db_state == INSTANCE_HA_STATE_CATCH_UP))) {
SendAlarmMsg(alarmIndex, logicClusterName, instanceName, ALM_AT_Resume);
} else {
SendAlarmMsg(alarmIndex, logicClusterName, instanceName, ALM_AT_Fault);
}
} else {
if (dnStatus->reportMsg.receive_status.peer_role != INSTANCE_ROLE_PENDING) {
SendAlarmMsg(alarmIndex, logicClusterName, instanceName, ALM_AT_Fault);
}
}
return;
}
void* DNStatusCheckMain(void *arg)
{
DnStatus dnStatus;
int32 i = *(int32*)arg;
errno_t rc;
pthread_t threadId = pthread_self();
uint32 dn_restart_count_check_time = 0;
uint32 dn_restart_count_check_time_in_hour = 0;
g_dnReportMsg[i].dnStatus.reportMsg.dn_restart_counts = 0;
uint32 check_dn_sql5_timer = g_check_dn_sql5_interval;
char* logicClusterName = NULL;
char instanceName[CM_NODE_NAME] = {0};
int alarmIndex = i;
write_runlog(LOG, "dn(%d) status check thread start, threadid %lu.\n", i, threadId);
int checkDummyTimes = CHECK_DUMMY_STATE_TIMES;
int ret = snprintf_s(instanceName, sizeof(instanceName), sizeof(instanceName) - 1,
"%s_%u", "dn", g_currentNode->datanode[i].datanodeId);
securec_check_intval(ret, (void)ret);
int32 running = PROCESS_UNKNOWN;
int index = -1;
AddThreadActivity(&index, threadId);
for (;;) {
set_thread_state(threadId);
struct stat instance_stat_buf = {0};
struct stat cluster_stat_buf = {0};
if (g_shutdownRequest || g_exitFlag || g_enableWalRecord) {
cm_sleep(5);
continue;
}
InitDNStatus(&dnStatus, i);
running = check_one_instance_status(GetDnProcessName(), g_currentNode->datanode[i].datanodeLocalDataPath, NULL);
if (g_currentNode->datanode[i].datanodeRole != DUMMY_STANDBY_DN) {
ret = DatanodeStatusCheck(&dnStatus, (uint32)i, running);
}
if (ret < 0 || g_currentNode->datanode[i].datanodeRole == DUMMY_STANDBY_DN) {
if (g_currentNode->datanode[i].datanodeRole != DUMMY_STANDBY_DN) {
write_runlog(ERROR, "DatanodeStatusCheck failed, ret=%d\n", ret);
}
if (g_currentNode->datanode[i].datanodeRole == DUMMY_STANDBY_DN &&
dnStatus.reportMsg.processStatus != INSTANCE_PROCESS_RUNNING && running != PROCESS_RUNNING) {
checkDummyTimes = CHECK_DUMMY_STATE_TIMES;
}
if (running == PROCESS_RUNNING) {
if (g_currentNode->datanode[i].datanodeRole == DUMMY_STANDBY_DN && checkDummyTimes > 0) {
checkDummyTimes--;
}
if (checkDummyTimes <= 0 || g_currentNode->datanode[i].datanodeRole != DUMMY_STANDBY_DN) {
dnStatus.reportMsg.processStatus = INSTANCE_PROCESS_RUNNING;
}
} else {
write_runlog(LOG, "set %u on offline.\n", dnStatus.reportMsg.instanceId);
char instance_manual_start_path[MAX_PATH_LEN] = {0};
dnStatus.reportMsg.processStatus = INSTANCE_PROCESS_DIED;
dnStatus.reportMsg.local_status.local_role = INSTANCE_ROLE_UNKNOWN;
rc = snprintf_s(instance_manual_start_path, MAX_PATH_LEN, MAX_PATH_LEN - 1,
"%s_%u",
g_cmInstanceManualStartPath,
g_currentNode->datanode[i].datanodeId);
securec_check_intval(rc, (void)rc);
if (stat(instance_manual_start_path, &instance_stat_buf) == 0 ||
stat(g_cmManualStartPath, &cluster_stat_buf) == 0 || !CheckStartDN()) {
dnStatus.reportMsg.local_status.db_state = INSTANCE_HA_STATE_MANUAL_STOPPED;
} else if (g_dnDiskDamage[i]) {
dnStatus.reportMsg.local_status.db_state = INSTANCE_HA_STATE_DISK_DAMAGED;
} else if (agentCheckPort(g_currentNode->datanode[i].datanodePort) > 0 ||
agentCheckPort(g_currentNode->datanode[i].datanodeLocalHAPort) > 0) {
dnStatus.reportMsg.local_status.db_state = INSTANCE_HA_STATE_PORT_USED;
} else {
* if instance is not running, cm_agent try to retsart it. if instance is still not running
* after MAX_INSTANCE_START times' trying, think it down.
*/
if (g_dnStartCounts[i] > max_instance_start) {
char build_pid_path[MAXPGPATH];
int rcs = snprintf_s(build_pid_path, MAXPGPATH, MAXPGPATH - 1,
"%s/gs_build.pid",
g_currentNode->datanode[i].datanodeLocalDataPath);
securec_check_intval(rcs, (void)rcs);
pgpid_t pid = get_pgpid(build_pid_path, MAXPGPATH);
if ((pid > 0 && !is_process_alive(pid)) || pid < 0) {
dnStatus.reportMsg.local_status.db_state = INSTANCE_HA_STATE_BUILD_FAILED;
} else {
dnStatus.reportMsg.local_status.db_state = INSTANCE_HA_STATE_UNKONWN;
}
} else {
dnStatus.reportMsg.local_status.db_state = INSTANCE_HA_STATE_STARTING;
}
}
dnStatus.reportMsg.local_status.buildReason = INSTANCE_HA_DATANODE_BUILD_REASON_UNKNOWN;
}
}
if (agent_backup_open == CLUSTER_STREAMING_STANDBY) {
ChangeLocalRoleInBackup(i, &dnStatus.reportMsg.local_status.local_role);
if (dnStatus.reportMsg.local_status.local_role == INSTANCE_ROLE_PRIMARY &&
dnStatus.reportMsg.local_status.db_state == INSTANCE_HA_STATE_NEED_REPAIR &&
dnStatus.reportMsg.local_status.buildReason == INSTANCE_HA_DATANODE_BUILD_REASON_DISCONNECT) {
ReportStreamingDRAlarm(ALM_AT_Fault, instanceName, alarmIndex, instanceName);
} else {
ReportStreamingDRAlarm(ALM_AT_Resume, instanceName, alarmIndex, NULL);
}
} else {
ReportStreamingDRAlarm(ALM_AT_Resume, instanceName, alarmIndex, NULL);
}
if (g_clusterType != V3SingleInstCluster &&
dnStatus.reportMsg.connectStatus == AGENT_TO_INSTANCE_CONNECTION_OK &&
g_currentNode->datanode[i].datanodeRole != DUMMY_STANDBY_DN) {
logicClusterName = get_logicClusterName_by_dnInstanceId(dnStatus.reportMsg.instanceId);
if (strcmp(g_agentEnableDcf, "on") != 0) {
if (dnStatus.reportMsg.local_status.local_role == INSTANCE_ROLE_PRIMARY) {
CheckAlmLocalPrimary(
&dnStatus, alarmIndex, (const char *)logicClusterName, (const char *)instanceName);
} else if (dnStatus.reportMsg.local_status.local_role == INSTANCE_ROLE_STANDBY) {
CheckAlmLocalStandby(
&dnStatus, alarmIndex, (const char *)logicClusterName, (const char *)instanceName);
}
}
}
dnStatus.reportMsg.dn_restart_counts = g_primaryDnRestartCounts[i];
if (dn_restart_count_check_time >= (DN_RESTART_COUNT_CHECK_TIME / agent_report_interval) ||
dnStatus.reportMsg.local_status.local_role == INSTANCE_ROLE_STANDBY) {
dn_restart_count_check_time = 0;
dnStatus.reportMsg.dn_restart_counts = 0;
g_primaryDnRestartCounts[i] = 0;
}
dnStatus.reportMsg.dn_restart_counts_in_hour = g_primaryDnRestartCountsInHour[i];
if (dn_restart_count_check_time_in_hour >= (DN_RESTART_COUNT_CHECK_TIME_HOUR / agent_report_interval) ||
dnStatus.reportMsg.local_status.local_role == INSTANCE_ROLE_STANDBY) {
dn_restart_count_check_time_in_hour = 0;
dnStatus.reportMsg.dn_restart_counts_in_hour = 0;
g_primaryDnRestartCountsInHour[i] = 0;
}
if (dnStatus.reportMsg.local_status.db_state == INSTANCE_HA_STATE_BUILD_FAILED) {
report_build_fail_alarm(ALM_AT_Fault, instanceName, i);
}
if (dnStatus.reportMsg.local_status.db_state == INSTANCE_HA_STATE_NORMAL ||
dnStatus.reportMsg.local_status.db_state == INSTANCE_HA_STATE_BUILDING) {
report_build_fail_alarm(ALM_AT_Resume, instanceName, i);
}
write_runlog(DEBUG1, "DatanodeStatusCheck: local role is %d, db state is %d, build reason is %d\n",
dnStatus.reportMsg.local_status.local_role, dnStatus.reportMsg.local_status.db_state,
dnStatus.reportMsg.local_status.buildReason);
DnCheckFloatIp(&dnStatus, (uint32)i, (bool8)(running == PROCESS_RUNNING));
(void)pthread_rwlock_wrlock(&(g_dnReportMsg[i].lk_lock));
rc = memcpy_s((void *)&(g_dnReportMsg[i].dnStatus.lpInfo), sizeof(AgentCmDnLocalPeer),
(void *)&dnStatus.lpInfo, sizeof(AgentCmDnLocalPeer));
securec_check_errno(rc, (void)rc);
rc = memcpy_s((void *)&(g_dnReportMsg[i].dnStatus.reportMsg), sizeof(agent_to_cm_datanode_status_report),
(void *)&dnStatus.reportMsg, sizeof(agent_to_cm_datanode_status_report));
securec_check_errno(rc, (void)rc);
rc = memcpy_s((void *)&(g_dnReportMsg[i].dnStatus.floatIpInfo), sizeof(CmaDnFloatIpInfo),
(void *)&dnStatus.floatIpInfo, sizeof(CmaDnFloatIpInfo));
securec_check_errno(rc, (void)rc);
(void)pthread_rwlock_unlock(&(g_dnReportMsg[i].lk_lock));
cm_sleep(agent_report_interval);
check_dn_sql5_timer = (check_dn_sql5_timer > 0) ? (check_dn_sql5_timer - 1) : g_check_dn_sql5_interval;
dn_restart_count_check_time++;
dn_restart_count_check_time_in_hour++;
UpdateThreadActivity(index);
}
}
void* WRFloatIpCheckMain(void *arg)
{
CmaWrFloatIp wrFloatIp;
int32 i = *(int32*)arg;
write_runlog(LOG, "WRFloatIpCheckMain: start to check wr float ip status.\n");
for (;;) {
if (g_shutdownRequest || !IsNeedCheckFloatIp() || (agent_backup_open != CLUSTER_PRIMARY) ||
!g_enableWalRecord) {
cm_sleep(SHUTDOWN_SLEEP_TIME);
continue;
}
InitWrFloatIp(&wrFloatIp, g_nodeId);
DnFloatIp *dnFloatIp = GetDnFloatIpByDnIdx(i);
wrFloatIp.count = dnFloatIp->dnFloatIpCount;
NetworkState state[MAX_FLOAT_IP_COUNT];
GetFloatIpNicStatus(dnFloatIp->instId, CM_INSTANCE_TYPE_DN, state, MAX_FLOAT_IP_COUNT);
for (uint32 k = 0; k < wrFloatIp.count; k++) {
wrFloatIp.netState[k] = state[k];
}
ReportCmsWrFloatIp(&wrFloatIp, g_nodeId);
NetworkOper oper = GetFloatIpOper(g_nodeId);
SetNicOper(dnFloatIp->instId, CM_INSTANCE_TYPE_DN, NETWORK_TYPE_FLOATIP, oper);
cm_sleep(agnet_report_wrFloatip_interval);
}
}
uint32 check_kerberos_state(const char* username)
{
if (check_one_instance_status("krb5kdc", "krb5kdc", NULL) == PROCESS_RUNNING) {
char actualCmd[MAX_PATH_LEN] = {0};
char kerberosCommandPath[MAX_PATH_LEN] = {0};
pid_t status;
int rcs = cmagent_getenv("KRB_HOME", kerberosCommandPath, sizeof(kerberosCommandPath));
if (rcs != EOK) {
write_runlog(LOG, "Get KRB_HOME failed, please check.\n");
return KERBEROS_STATUS_UNKNOWN;
} else {
check_input_for_security(kerberosCommandPath);
int32 ret = snprintf_s(actualCmd, MAX_PATH_LEN, MAX_PATH_LEN - 1,
"%s/bin/kinit -k -t %s/kerberos/%s.keytab %s/opengauss.org@OPENGAUSS.ORG",
kerberosCommandPath, kerberosCommandPath, username, username);
securec_check_intval(ret, (void)ret);
check_input_for_security(actualCmd);
status = system(actualCmd);
if (status == -1) {
write_runlog(ERROR, "fail to execute command %s, and errno=%d.", actualCmd, errno);
return KERBEROS_STATUS_UNKNOWN;
} else {
if (WIFEXITED(status)) {
if (WEXITSTATUS(status) == 0) {
return KERBEROS_STATUS_NORMAL;
} else {
return KERBEROS_STATUS_ABNORMAL;
}
} else {
return KERBEROS_STATUS_UNKNOWN;
}
}
}
} else {
return KERBEROS_STATUS_DOWN;
}
}
void get_kerberosConfigFile_info(const char* kerberosConfigFile, char* kerberosIp, uint32* kerberosPort, int* roleFlag)
{
FILE* kerberos_config_fd;
char buff[MAX_BUFF] = {0};
char validstring[MAX_BUFF] = {0};
errno_t rc;
int rcs;
char ip[CM_IP_LENGTH] = {0};
int port = 0;
if ((kerberos_config_fd = fopen(kerberosConfigFile, "re")) != NULL) {
while (!feof(kerberos_config_fd)) {
if (fgets(buff, MAX_BUFF, kerberos_config_fd) == NULL) {
write_runlog(DEBUG1, "kerberos ip and port unkonw !\n");
break;
}
if (strstr(buff, "kdc ") != NULL || strstr(buff, "kdc=") != NULL) {
rcs = sscanf_s(buff, "%[^1-9]%[^:]%*c%d", validstring, MAX_BUFF, ip, CM_IP_LENGTH, &port);
check_sscanf_s_result(rcs, 3);
securec_check_intval(rcs, (void)fclose(kerberos_config_fd));
if (strcmp(ip, g_currentNode->backIps[0]) == 0 && port != 0) {
rc = strncpy_s(kerberosIp, CM_IP_LENGTH, ip, strlen(ip));
securec_check_errno(rc, (void)fclose(kerberos_config_fd));
*kerberosPort = (uint32)port;
break;
}
*roleFlag = *roleFlag + 1;
}
}
if (port == 0) {
write_runlog(LOG, "Please reinstall kerberos!\n");
}
(void)fclose(kerberos_config_fd);
return;
} else {
write_runlog(LOG, "kerberos config open error !\n");
return;
}
}
void* KerberosStatusCheckMain(void* arg)
{
agent_to_cm_kerberos_status_report report_msg;
errno_t rc;
pthread_t threadId = pthread_self();
set_thread_state(threadId);
char kerberos_config_path[MAX_PATH_LEN] = {0};
write_runlog(LOG, "kerberos status check thread start, threadid %lu.\n", threadId);
char kerberosIp[CM_IP_LENGTH] = {0};
uint32 kerberosPort = 0;
int roleFlag = 0;
rc = memset_s(
&report_msg, sizeof(agent_to_cm_kerberos_status_report), 0, sizeof(agent_to_cm_kerberos_status_report));
securec_check_errno(rc, (void)rc);
report_msg.msg_type = MSG_AGENT_CM_KERBEROS_STATUS;
int isKerberos = cmagent_getenv("MPPDB_KRB5_FILE_PATH", kerberos_config_path, sizeof(kerberos_config_path));
if (isKerberos != EOK) {
write_runlog(DEBUG1, "KerberosStatusCheckMain: MPPDB_KRB5_FILE_PATH get fail.\n");
return NULL;
}
struct stat stat_buf = {0};
check_input_for_security(kerberos_config_path);
canonicalize_path(kerberos_config_path);
if (stat(kerberos_config_path, &stat_buf) != 0) {
write_runlog(DEBUG1, "KerberosStatusCheckMain: kerberos config file not exist.\n");
return NULL;
}
report_msg.node = g_currentNode->node;
rc = strncpy_s(report_msg.nodeName, MAXLEN, g_currentNode->nodeName, strlen(g_currentNode->nodeName));
securec_check_errno(rc, (void)rc);
get_kerberosConfigFile_info(kerberos_config_path, kerberosIp, &kerberosPort, &roleFlag);
rc = strncpy_s(report_msg.kerberos_ip, CM_IP_LENGTH, kerberosIp, strlen(kerberosIp));
securec_check_errno(rc, (void)rc);
report_msg.port = kerberosPort;
if (roleFlag == 0) {
rc = strncpy_s(report_msg.role, MAXLEN, "Primary", strlen("Primary"));
securec_check_errno(rc, (void)rc);
} else {
rc = strncpy_s(report_msg.role, MAXLEN, "Standby", strlen("Standby"));
securec_check_errno(rc, (void)rc);
}
for (;;) {
if (g_shutdownRequest) {
cm_sleep(5);
continue;
}
report_msg.status = check_kerberos_state(pw->pw_name);
int ret = 0;
if (report_msg.status == KERBEROS_STATUS_ABNORMAL || report_msg.status == KERBEROS_STATUS_DOWN) {
kill_instance_force("krb5kdc", INSTANCE_KERBEROS);
cm_sleep(1);
ret = system("krb5kdc");
if (ret != 0) {
write_runlog(ERROR, "run krb5kdc command failed and restart fail, errno=%d.!\n", errno);
}
}
(void)pthread_rwlock_wrlock(&(g_kerberosReportMsg.lk_lock));
rc = memcpy_s((void*)&(g_kerberosReportMsg.report_msg),
sizeof(agent_to_cm_kerberos_status_report),
(void*)&report_msg,
sizeof(agent_to_cm_kerberos_status_report));
securec_check_errno(rc, (void)rc);
(void)pthread_rwlock_unlock(&(g_kerberosReportMsg.lk_lock));
cm_sleep(g_agentKerberosStatusCheckInterval);
}
}
void CheckSharedDiskUsage(uint32 &vgdataPathUsage, uint32 &vglogPathUsage)
{
FILE *fp;
char result[1024];
double percent1 = 0.0, percent2 = 0.0;
fp = popen("dsscmd lsvg | awk 'NR==2 || NR==3 {print $NF}'", "r");
if (fp == NULL) {
write_runlog(ERROR, "Failed to exec command(dsscmd lsvg).\n");
vgdataPathUsage = 0;
vglogPathUsage = 0;
return;
}
if (fgets(result, sizeof(result)-1, fp) != NULL) {
sscanf(result, "%lf", &percent1);
}
if (fgets(result, sizeof(result)-1, fp) != NULL) {
sscanf(result, "%lf", &percent2);
}
if (percent1 > 0 && percent1 < 1) {
percent1 = 1;
}
if (percent2 > 0 && percent2 < 1) {
percent2 = 1;
}
vgdataPathUsage = (uint)percent1;
vglogPathUsage = (uint)percent2;
pclose(fp);
}
* @brief Get DN node log path disk usage and datapath disk usage, send them to the CMS
*
*/
void CheckDiskForDNDataPath()
{
for (uint32 ii = 0; ii < g_currentNode->datanodeCount; ii++) {
AgentToCmDiskUsageStatusReport status;
status.msgType = (int)MSG_AGENT_CM_DISKUSAGE_STATUS;
status.instanceId = g_currentNode->datanode[ii].datanodeId;
status.logPathUsage = CheckDiskForLogPath();
uint32 dataPathUsage = GetDiskUsageForPath(g_currentNode->datanode[ii].datanodeLocalDataPath);
uint32 linkPathUsage = GetDiskUsageForLinkPath(g_currentNode->datanode[ii].datanodeLocalDataPath);
status.dataPathUsage = (dataPathUsage > linkPathUsage) ? dataPathUsage : linkPathUsage;
status.readOnly = g_dnReadOnly[ii];
status.instanceType = INSTANCE_TYPE_DATANODE;
if (IsCusResExistLocal()) {
CheckSharedDiskUsage(status.vgdataPathUsage, status.vglogPathUsage);
write_runlog(DEBUG1, "vgdataPathUsage:%u, vglogPathUsage:%u.\n",
status.vgdataPathUsage, status.vglogPathUsage);
} else {
status.vgdataPathUsage = 0;
status.vglogPathUsage = 0;
}
write_runlog(DEBUG1, "[%s] msgType:%d, instanceId:%u, logPathUsage:%u, linkPathUsage: %u, dataPathUsage:%u.\n",
__FUNCTION__, status.msgType, status.instanceId, status.logPathUsage, linkPathUsage, status.dataPathUsage);
(void)pthread_rwlock_wrlock(&(g_dnReportMsg[ii].lk_lock));
errno_t rc = memcpy_s((void *)&(g_dnReportMsg[ii].dnStatus.diskUsageMsg),
sizeof(AgentToCmDiskUsageStatusReport), (void *)&status, sizeof(AgentToCmDiskUsageStatusReport));
securec_check_errno(rc, (void)rc);
(void)pthread_rwlock_unlock(&(g_dnReportMsg[ii].lk_lock));
}
}
* This function is used to ensure whether the status modify;
*/
static int IsClusterInRedoState()
{
FILE *fp;
char buffer[1024] = {0};
const char* expectedString = "in on-demand redo";
char *foundString = NULL;
fp = popen("pg_controldata +data | grep 'Cluster status:' | "
"awk '{for (i=NF-2;i<=NF;i++) printf \"%s \", $i; printf \"\\n\"}'", "r");
if (fp == NULL) {
write_runlog(LOG, "Failed to exec command(pg_controldata +data).\n");
return UNEXPECT_ONDEMAND_RECOVERY;
}
if (fgets(buffer, sizeof(buffer), fp) == NULL) {
pclose(fp);
return UNEXPECT_ONDEMAND_RECOVERY;
}
pclose(fp);
foundString = buffer;
char *end = foundString + strlen(foundString) - 1;
while (end >= foundString && (*end == '\n' || *end == '\r' || *end == '\t' || *end == ' ')) {
*end = '\0';
end--;
}
if (*foundString == '\0') {
return UNEXPECT_ONDEMAND_RECOVERY;
}
if (!strcmp(foundString, expectedString)) {
return IN_ONDEMAND_RECOVERY;
} else {
return NOT_IN_ONDEMAND_RECOVERY;
}
}
static void AgentSendOnDemandStatus(int onDemandStatus)
{
agent_to_cm_ondemand_status_report reportMsg = {0};
reportMsg.msg_type = (int)MSG_AGENT_ONDEMAND_STATUES_REPORT;
reportMsg.onDemandStatus = onDemandStatus;
reportMsg.reportTime = time(NULL);
reportMsg.nodeId = g_nodeHeader.node;
PushMsgToCmsSendQue((char *)&reportMsg, (uint32)sizeof(agent_to_cm_ondemand_status_report), "ondemand status");
write_runlog(DEBUG1, "Node(%u) ondemand status send to cms, status is: %d\n", reportMsg.nodeId, onDemandStatus);
}
* This function is main entry for pgdata charge.
* It can be extended in the future.
*/
void PGDataControlCheck()
{
int clusterStatusUnderOnDemand = IsClusterInRedoState();
AgentSendOnDemandStatus(clusterStatusUnderOnDemand);
}
static void PingPeerIP(int* count, const char localIP[CM_IP_LENGTH], const char peerIP[CM_IP_LENGTH])
{
char command[MAXPGPATH] = {0};
char buf[MAXPGPATH];
int rc;
uint32 tryTimes = 3;
const char *pingStr = GetPingStr(GetIpVersion(peerIP));
rc = snprintf_s(command, MAXPGPATH, MAXPGPATH - 1,
"%s -c 1 -w 1 -I %s %s > /dev/null;if [ $? == 0 ];then echo success;else echo fail;fi;",
pingStr, localIP, peerIP);
securec_check_intval(rc, (void)rc);
write_runlog(DEBUG1, "ping command is: %s.\n", command);
while (tryTimes > 0) {
FILE* fp = popen(command, "r");
if (fp == NULL) {
write_runlog(ERROR, "popen failed\n.");
return;
}
if (fgets(buf, sizeof(buf), fp) != NULL) {
if (strstr(buf, "success") != NULL) {
(*count)++;
(void)pclose(fp);
return;
}
}
cm_sleep(1);
tryTimes--;
(void)pclose(fp);
}
write_runlog(ERROR, "ping peer ip failed: %s, command is: %s\n.", buf, command);
return;
}
void GetPingSuccessCount(int i, int* count)
{
if (g_multi_az_cluster) {
for (uint32 j = 0; j < g_dn_replication_num - 1; ++j) {
PingPeerIP(count, g_currentNode->datanode[i].datanodeLocalHAIP[0],
g_currentNode->datanode[i].peerDatanodes[j].datanodePeerHAIP[0]);
}
} else {
PingPeerIP(count, g_currentNode->datanode[i].datanodeLocalHAIP[0],
g_currentNode->datanode[i].datanodePeerHAIP[0]);
PingPeerIP(count, g_currentNode->datanode[i].datanodeLocalHAIP[0],
g_currentNode->datanode[i].datanodePeer2HAIP[0]);
}
}
void CheckDNConnectionStatus(int i, int alarmIndex, const char *instanceName)
{
AlarmType alarmType = ALM_AT_Resume;
if (!g_mostAvailableSync[i]) {
int count = 0;
GetPingSuccessCount(i, &count);
if (count == 0) {
write_runlog(LOG, "dn(%u) is disconnected from other dn.\n", g_currentNode->datanode[i].datanodeId);
g_dnPingFault[i] = true;
if (g_dnReportMsg[i].dnStatus.reportMsg.local_status.local_role == INSTANCE_ROLE_PRIMARY &&
!g_isPauseArbitration) {
if (g_enableWalRecord) {
for (uint32 i = 0; i < GetLocalResConfCount(); ++i) {
if (strcmp(g_resConf[i].resName, "gr") == 0) {
ManualStopLocalResInst(&g_resConf[i]);
}
}
} else {
immediate_stop_one_instance(g_currentNode->datanode[i].datanodeLocalDataPath, INSTANCE_DN);
}
}
alarmType = ALM_AT_Fault;
} else {
g_dnPingFault[i] = false;
}
} else {
g_dnPingFault[i] = false;
}
ReportDNDisconnectAlarm(alarmType, instanceName, alarmIndex);
}
void* DNConnectionStatusCheckMain(void *arg)
{
int i = *(int*)arg;
int alarmIndex = i;
char instanceName[CM_NODE_NAME] = {0};
pthread_t threadId = pthread_self();
if ((i < 0) || (i >= CM_MAX_DATANODE_PER_NODE)) {
write_runlog(ERROR, "DN index [%d] is invalid, failed to start DNConnection thread.\n", i);
return NULL;
}
if (g_single_node_cluster || IsBoolCmParamTrue(g_agentEnableDcf)) {
write_runlog(LOG, "instanceId(%u) is single node cluster or in dcf mode, no need connection status check.\n",
g_currentNode->datanode[i].datanodeId);
return NULL;
}
write_runlog(LOG, "dn(%d) connection status check thread start, threadid %lu.\n", i, threadId);
int ret = snprintf_s(instanceName, sizeof(instanceName), sizeof(instanceName) - 1,
"%s_%u", "dn", g_currentNode->datanode[i].datanodeId);
securec_check_intval(ret, (void)ret);
for (;;) {
set_thread_state(threadId);
if (g_shutdownRequest || g_enableWalRecord) {
cm_sleep(5);
continue;
}
CheckDNConnectionStatus(i, alarmIndex, instanceName);
cm_sleep(agent_report_interval);
}
}
static bool IsDeviceNameSame(const char *device, int deviceCount, char * const *deviceName)
{
for (int i = 0; i < deviceCount; ++i) {
if (strcmp(device, deviceName[i]) == 0) {
return true;
}
}
return false;
}
static char **GetAllDisk(int &deviceCount)
{
char **result;
char tmpName[MAX_DEVICE_DIR] = {0};
errno_t rc;
size_t resultLen = (g_currentNode->datanodeCount + 1) * sizeof(char*);
result = (char**)malloc(resultLen);
if (result == NULL) {
write_runlog(ERROR, "[CmReadfile] malloc failed, out of memory.\n");
return NULL;
}
rc = memset_s(result, resultLen, 0, resultLen);
securec_check_errno(rc, (void)rc);
deviceCount = 0;
if (g_currentNode->coordinate == 1) {
CmGetDisk(g_currentNode->DataPath, tmpName, MAX_DEVICE_DIR);
result[deviceCount] = strdup(tmpName);
if (result[deviceCount] == NULL) {
write_runlog(ERROR, "out of memory, deviceCount = %d\n", deviceCount);
return NULL;
}
++deviceCount;
rc = memset_s(tmpName, MAX_DEVICE_DIR, 0, MAX_DEVICE_DIR);
securec_check_errno(rc, (void)rc);
}
for (uint32 i = 0; i < g_currentNode->datanodeCount; ++i) {
CmGetDisk(g_currentNode->datanode[i].datanodeLocalDataPath, tmpName, MAX_DEVICE_DIR);
if (!IsDeviceNameSame(tmpName, deviceCount, result)) {
result[deviceCount] = strdup(tmpName);
if (result[deviceCount] == NULL) {
write_runlog(ERROR, "out of memory, deviceCount = %d\n", deviceCount);
return NULL;
}
++deviceCount;
}
rc = memset_s(tmpName, MAX_DEVICE_DIR, 0, MAX_DEVICE_DIR);
securec_check_errno(rc, (void)rc);
}
return result;
}
static bool IsSymbolRight(const char *str)
{
int count = 0;
for (int i = 0; str[i] != '\0'; ++i) {
if (str[i] == ',') {
++count;
}
}
if (count == THRESHOLD_FORMAT) {
return true;
}
return false;
}
static bool IsValueRight(const char *value, int ¶m)
{
if (value == NULL) {
write_runlog(ERROR, "threshold value = NULL.\n");
return false;
}
if (CM_is_str_all_digit(value) != 0) {
write_runlog(ERROR, "threshold value = %s, is wrong.\n", value);
return false;
}
param = (int)strtol(value, NULL, DECIMAL_NOTATION);
if (param < THRESHOLD_MIN_VALUE || param > THRESHOLD_MAX_VALUE) {
write_runlog(ERROR, "threshold value = %s, out of range.\n", value);
return false;
}
return true;
}
static status_t GetThreshold(EnvThreshold &threshold)
{
char *pLeft = NULL;
char *pValue;
char envStr[CM_PATH_LENGTH] = {0};
if (strcmp(g_environmentThreshold, "") == 0) {
write_runlog(DEBUG1, "environment_threshold is NULL.\n");
return CM_ERROR;
}
errno_t rc = strcpy_s(envStr, CM_PATH_LENGTH, g_environmentThreshold);
securec_check_errno(rc, (void)rc);
char *tmp = trim(envStr);
write_runlog(DEBUG1, "environment threshold, tmp=%s.\n", tmp);
if (tmp[strlen(tmp) - 1] == ')') {
tmp[strlen(tmp) - 1] = '\0';
} else {
write_runlog(ERROR, "line:%d, environment threshold format is wrong.\n", __LINE__);
return CM_ERROR;
}
if (tmp[0] == '(') {
tmp++;
} else {
write_runlog(ERROR, "line:%d, environment threshold format is wrong.\n", __LINE__);
return CM_ERROR;
}
if (!IsSymbolRight(tmp)) {
write_runlog(ERROR, "line:%d, environment threshold format is wrong.\n", __LINE__);
return CM_ERROR;
}
pValue = strtok_r(tmp, ",", &pLeft);
if (!IsValueRight(pValue, threshold.mem)) {
return CM_ERROR;
}
pValue = strtok_r(NULL, ",", &pLeft);
if (!IsValueRight(pValue, threshold.cpu)) {
return CM_ERROR;
}
pValue = strtok_r(NULL, ",", &pLeft);
if (!IsValueRight(pValue, threshold.disk)) {
return CM_ERROR;
}
pValue = strtok_r(NULL, ",", &pLeft);
if (!IsValueRight(pValue, threshold.instMem)) {
return CM_ERROR;
}
if (!IsValueRight(pLeft, threshold.instPool)) {
return CM_ERROR;
}
return CM_SUCCESS;
}
static int CheckMemoryHave()
{
FILE *fp;
uint64 memFree = 0;
uint64 memTotal = 0;
int ret;
bool haveGetFree = false;
bool haveGetTotal = false;
char line[CM_PATH_LENGTH] = {0};
if ((fp = fopen(FILE_MEMINFO, "re")) == NULL) {
write_runlog(ERROR, "failed to open file %s.\n", FILE_MEMINFO);
return -1;
}
while (fgets(line, CM_PATH_LENGTH, fp) != NULL) {
if (strncmp(line, "MemTotal", strlen("MemTotal")) == 0) {
ret = sscanf_s(line, "MemTotal: %lu kB\n", &memTotal);
check_sscanf_s_result(ret, 1);
securec_check_intval(ret, (void)ret);
write_runlog(DEBUG1, "CheckMemoryHave memTotal = %lu.\n", memTotal);
haveGetTotal = true;
}
if (strncmp(line, "MemFree", strlen("MemFree")) == 0) {
ret = sscanf_s(line, "MemFree: %lu kB", &memFree);
check_sscanf_s_result(ret, 1);
securec_check_intval(ret, (void)ret);
write_runlog(DEBUG1, "CheckMemoryHave memTotal = %lu.\n", memFree);
haveGetFree = true;
}
if (haveGetTotal && haveGetFree) {
break;
}
}
(void)fclose(fp);
if (memTotal == 0) {
write_runlog(ERROR, "get memTotal(%lu) info is 0.\n", memTotal);
return -1;
}
return (int)(PERCENT - (memFree * PERCENT / memTotal));
}
static int CheckCpuHave()
{
return ReadCpuStatus(1, NULL, true);
}
static void CheckDiskIoHave(const char *deviceName, int disk)
{
int cpuNum;
int diskIoHave;
static IoStat ioStatus = {0};
if (deviceName == NULL) {
write_runlog(LOG, "device name is NULL, can't check its disk IO.\n");
return;
}
cpuNum = GetCpuCount();
diskIoHave = (int)ReadDiskIOStat(deviceName, cpuNum, &ioStatus, false);
if (diskIoHave > disk) {
write_runlog(LOG, "{\"CMA disk IO is more than threshold\":"
"{\"disk IO\":{\"name\":\"%s\",\"actual\":\"%d%%\", \"threshold\":\"%d%%\"}}}\n",
deviceName, diskIoHave, disk);
}
return;
}
static void CheckSysStatus(const EnvThreshold &threshold)
{
int memHave = 0;
int cpuHave = 0;
bool isOverflow;
if (threshold.mem == 0 && threshold.cpu == 0) {
write_runlog(DEBUG5, "threshold mem and cpu is 0, not need do check.\n");
return;
}
if (threshold.mem != 0) {
if ((memHave = CheckMemoryHave()) < 0) {
write_runlog(ERROR, "get memory info fail.\n");
return;
}
}
if (threshold.cpu != 0) {
if ((cpuHave = CheckCpuHave()) < 0) {
write_runlog(ERROR, "get cpu info fail.\n");
return;
}
}
isOverflow = (memHave > threshold.mem) || (cpuHave > threshold.cpu);
if (isOverflow) {
write_runlog(LOG, "{\"CMA physical resource is more than threshold\":"
"{\"memory\":{\"actual\":\"%d%%\",\"threshold\":\"%d%%\"},"
"\"CPU\":{\"actual\":\"%d%%\",\"threshold\":\"%d%%\"}}}\n",
memHave, threshold.mem, cpuHave, threshold.cpu);
}
return;
}
static void CheckDiskStatus(const EnvThreshold &threshold, const char * const *deviceName, int deviceCount)
{
if (threshold.disk == 0) {
write_runlog(DEBUG5, "threshold disk is 0, not need do check.\n");
return;
}
for (int i = 0; i < deviceCount; ++i) {
CheckDiskIoHave(deviceName[i], threshold.disk);
}
return;
}
void *CheckNodeStatusThreadMain(void * const arg)
{
int deviceCount = 0;
long expiredTime;
struct timeval checkEnd;
struct timeval checkBegin;
EnvThreshold threshold = {0};
char **deviceName = GetAllDisk(deviceCount);
if (deviceName == NULL) {
write_runlog(ERROR, "CheckNodeStatusThreadMain, out of memory.\n");
return NULL;
}
write_runlog(LOG, "CMA deviceCount = %d.\n", deviceCount);
for (;;) {
if (g_shutdownRequest) {
cm_sleep(SHUTDOWN_SLEEP_TIME);
continue;
}
(void)gettimeofday(&checkBegin, NULL);
if (GetThreshold(threshold) != CM_SUCCESS) {
threshold = {0, 0, 0, 0, 0};
}
CheckSysStatus(threshold);
CheckDiskStatus(threshold, deviceName, deviceCount);
#ifdef ENABLE_MULTIPLE_NODES
CheckAllInstStatus(&threshold);
#endif
(void)gettimeofday(&checkEnd, NULL);
expiredTime = (checkEnd.tv_sec - checkBegin.tv_sec);
write_runlog(DEBUG5, "CheckNodeStatusThreadMain take %ld seconds.\n", expiredTime);
if (expiredTime < CHECK_INTERVAL) {
cm_sleep((unsigned int)(CHECK_INTERVAL - expiredTime));
}
}
return NULL;
}
int CreateCheckNodeStatusThread()
{
int err;
pthread_t thrId;
if ((err = pthread_create(&thrId, NULL, CheckNodeStatusThreadMain, NULL)) != 0) {
write_runlog(ERROR, "Failed to create new thread: error %d.\n", err);
return err;
}
return 0;
}
void InitSystemStatInfo(SystemStatInfo* systemStat)
{
errno_t rc = memset_s(systemStat, sizeof(SystemStatInfo), 0, sizeof(SystemStatInfo));
securec_check_errno(rc, (void)rc);
systemStat->memoryStatInfo.lastReportTime = GetMonotonicTimeS();
systemStat->cpuStatInfo.lastReportTime = GetMonotonicTimeS();
int diskCount = 0;
char **diskName = GetAllDisk(diskCount);
if (diskName == NULL || diskCount <= 0) {
write_runlog(ERROR, "Get disk info failed.\n");
return;
}
systemStat->diskStatInfo.diskCount = diskCount;
systemStat->diskStatInfo.disIoStatInfo = new DisIoStatInfo[diskCount];
if (systemStat->diskStatInfo.disIoStatInfo == NULL) {
write_runlog(ERROR, "InitSystemStatInfo: out of memory, diskCount = %d\n", diskCount);
for (int i = 0; i < diskCount; ++i) {
free(diskName[i]);
}
free(diskName);
return;
}
for (int i = 0; i < diskCount; ++i) {
if (CM_IS_EMPTY_STR(diskName[i])) {
continue;
}
rc = strcpy_s(systemStat->diskStatInfo.disIoStatInfo[i].diskName, MAX_DEVICE_DIR, diskName[i]);
securec_check_errno(rc, (void)rc);
systemStat->diskStatInfo.disIoStatInfo[i].lastReportTime = GetMonotonicTimeS();
systemStat->diskStatInfo.disIoStatInfo[i].ioUtil = 0.0f;
systemStat->diskStatInfo.disIoStatInfo[i].svctm = 0.0f;
systemStat->diskStatInfo.disIoStatInfo[i].lastReportIoUtil = 0.0f;
systemStat->diskStatInfo.disIoStatInfo[i].totalWeight = 0;
systemStat->diskStatInfo.disIoStatInfo[i].lastReadCount = 0;
systemStat->diskStatInfo.disIoStatInfo[i].lastWriteCount = 0;
systemStat->diskStatInfo.disIoStatInfo[i].lastCheckTime = 0;
systemStat->diskStatInfo.disIoStatInfo[i].lastIoTime = 0;
systemStat->diskStatInfo.disIoStatInfo[i].weightWindow = std::deque<uint32>();
write_runlog(LOG, "start check device(%s).\n", diskName[i]);
}
for (int i = 0; i < diskCount; ++i) {
free(diskName[i]);
}
free(diskName);
}
status_t SetMemeryStatInfo(const char* line, MemoryStatInfo* memoryStatInfo, uint32& count)
{
if (CM_IS_EMPTY_STR(line)) {
return CM_ERROR;
}
uint32 length = ELEMENT_COUNT(g_memoryCheckInfoList);
for (uint32 i = 0; i < length; i++) {
MemCheckInfo memCheckInfo = g_memoryCheckInfoList[i];
if (strncmp(line, memCheckInfo.name, strlen(memCheckInfo.name)) != 0) {
continue;
}
int32 ret = sscanf_s(line, memCheckInfo.info, &(memoryStatInfo->memItemList[memCheckInfo.item]));
check_sscanf_s_result(ret, 1);
if (ret == -1) {
write_runlog(ERROR, "failed to sscanf %s, line=[%s].\n", memCheckInfo.info, line);
return CM_ERROR;
}
++count;
return CM_SUCCESS;
}
return CM_SUCCESS;
}
status_t CheckMemoryStatus(MemoryStatInfo* memoryStatInfo)
{
char line[CM_PATH_LENGTH] = {0};
uint32 success = 0;
FILE* fp = fopen(FILE_MEMINFO, "re");
if (fp == NULL) {
write_runlog(ERROR, "failed to open file %s.\n", FILE_MEMINFO);
return CM_ERROR;
}
while (fgets(line, CM_PATH_LENGTH, fp)!= NULL) {
if (SetMemeryStatInfo(line, memoryStatInfo, success) != CM_SUCCESS) {
FCLOSE_AND_RESET(fp);
return CM_ERROR;
}
if (success == (uint32)MEM_STAT_BUTT) {
break;
}
}
fclose(fp);
uint64 sysMemUsed = memoryStatInfo->memItemList[MEM_STAT_TOTAL] - memoryStatInfo->memItemList[MEM_STAT_FREE]-
memoryStatInfo->memItemList[MEM_STAT_BUFFERS] - memoryStatInfo->memItemList[MEM_STAT_CACHED];
uint64 appMemUsed = memoryStatInfo->memItemList[MEM_STAT_TOTAL] - memoryStatInfo->memItemList[MEM_STAT_AVAILABLE];
if (sysMemUsed > memoryStatInfo->memItemList[MEM_STAT_TOTAL]
|| appMemUsed > memoryStatInfo->memItemList[MEM_STAT_TOTAL]) {
write_runlog(ERROR, "check memory status failed, memTotal(%lu), memSys(%lu), memApp(%lu).\n",
memoryStatInfo->memItemList[MEM_STAT_TOTAL], sysMemUsed, appMemUsed);
return CM_ERROR;
}
memoryStatInfo->systemMemUsedUtil = (float)sysMemUsed /
(float)memoryStatInfo->memItemList[MEM_STAT_TOTAL] * PERCENT;
memoryStatInfo->appMemUsedUtil = (float)appMemUsed / (float)memoryStatInfo->memItemList[MEM_STAT_TOTAL] * PERCENT;
return CM_SUCCESS;
}
void ReportSystemStatusAlarm(const SystemStatInfo* systemStat, const EnvThreshold* threshold)
{
int memUsed = (int)systemStat->memoryStatInfo.systemMemUsedUtil;
int appMemUsed = (int)systemStat->memoryStatInfo.appMemUsedUtil;
if (threshold->mem > 0 && (memUsed > threshold->mem || appMemUsed > threshold->mem)) {
write_runlog(LOG, "system memory usage is %u, app memory usage is %u, threshold is %u, report alarm.\n",
memUsed, appMemUsed, threshold->mem);
ReportMemoryAbnormalAlarm(memUsed, appMemUsed, threshold->mem);
}
int cpuUsed = (int)systemStat->cpuStatInfo.cpuUtil;
if (threshold->cpu > 0 && cpuUsed > threshold->cpu) {
write_runlog(LOG, "system cpu usage is %u, threshold is %u, report alarm.\n", cpuUsed, threshold->cpu);
ReportCpuAbnormalAlarm(cpuUsed, threshold->cpu);
}
for (uint32 i = 0; i < systemStat->diskStatInfo.diskCount; ++i) {
DisIoStatInfo* disIoStatInfo = &(systemStat->diskStatInfo.disIoStatInfo[i]);
if (CM_IS_EMPTY_STR(disIoStatInfo->diskName)) {
continue;
}
int ioUtil = (int)disIoStatInfo->ioUtil;
if (threshold->disk > 0 && ioUtil > threshold->disk) {
write_runlog(LOG, "system disk(%s) io usage is %u, threshold is %u, report alarm.\n",
disIoStatInfo->diskName,
ioUtil,
threshold->disk);
ReportDiskIOAbnormalAlarm(disIoStatInfo->diskName, ioUtil, threshold->disk);
}
float svctm = disIoStatInfo->svctm;
for (int j = SVCTM_LEVEL_CEIL - 1; j >= 0; --j) {
if (svctm >= g_slowIoLevelInfoList[j].threshold) {
disIoStatInfo->weightWindow.push_back(g_slowIoLevelInfoList[j].weight);
disIoStatInfo->totalWeight += g_slowIoLevelInfoList[j].weight;
break;
}
}
AlarmType alarmType = ALM_AT_Resume;
if (disIoStatInfo->weightWindow.size() > SLOW_DISK_CHECK_PERIOD) {
disIoStatInfo->totalWeight -= disIoStatInfo->weightWindow.front();
disIoStatInfo->weightWindow.pop_front();
char details[MAX_PATH_LEN] = {0};
if (disIoStatInfo->totalWeight > SLOW_DISK_CHECK_THRESHOLD) {
alarmType = ALM_AT_Fault;
int ret = snprintf_s(details, MAX_PATH_LEN, MAX_PATH_LEN - 1, "weight: %lu, period: %lu, "
"threshold: %lu", disIoStatInfo->totalWeight, SLOW_DISK_CHECK_PERIOD, SLOW_DISK_CHECK_THRESHOLD);
securec_check_intval(ret, (void)ret);
write_runlog(LOG, "detected slow disk(%s), details: %s, report alarm.\n",
disIoStatInfo->diskName,
details);
}
ReportSlowDiskAlarm(disIoStatInfo->diskName, alarmType, i, details);
}
}
}
status_t SetCpuStatInfo(const char* line, CpuSimpleInfo* cpuSimpleInfo)
{
const int cpuNameLen = 8;
const int cpuInfoCount = 11;
char cpuName[cpuNameLen] = {0};
CpuInfo* tmpCpuInfo = &(cpuSimpleInfo->cpuInfo);
errno_t rc = sscanf_s(line, "%s %lu %lu %lu %lu %lu %lu %lu %lu %lu %lu", cpuName, cpuNameLen,
&tmpCpuInfo->cpuUser, &tmpCpuInfo->cpuNice, &tmpCpuInfo->cpuSys, &tmpCpuInfo->cpuIdle, &tmpCpuInfo->cpuIwait,
&tmpCpuInfo->cpuHardirq, &tmpCpuInfo->cpuSoftirq, &tmpCpuInfo->cpuSteal, &tmpCpuInfo->cpuGuest,
&tmpCpuInfo->cpuGuestNice);
check_sscanf_s_result(rc, cpuInfoCount);
if (rc != cpuInfoCount) {
write_runlog(ERROR, "get cpu info from %s failed, line %s, count %d.\n", FILE_CPUSTAT, line, rc);
return CM_ERROR;
}
cpuSimpleInfo->cpuIdleTime = tmpCpuInfo->cpuIdle;
cpuSimpleInfo->cpuUserTime = tmpCpuInfo->cpuUser;
cpuSimpleInfo->cpuSystemTime = tmpCpuInfo->cpuSys;
cpuSimpleInfo->cpuTotalTime = 0;
uint64* value = (uint64*)tmpCpuInfo;
uint32 len = (uint32)sizeof(CpuInfo) / sizeof(uint64);
for (uint32 i = 0; i < len; ++i) {
cpuSimpleInfo->cpuTotalTime += *(value++);
}
return (strcmp(cpuName, "cpu") == 0) ? CM_SUCCESS : CM_ERROR;
}
status_t CheckCpuStatus(CpuStatInfo* cpuStatInfo)
{
CpuSimpleInfo curCpuInfo = {0};
char line[MAX_PATH_LEN] = {0};
FILE* fp = fopen(FILE_CPUSTAT, "re");
if (fp == NULL) {
write_runlog(ERROR, "failed to open file %s.\n", FILE_CPUSTAT);
return CM_ERROR;
}
while (fgets(line, sizeof(line), fp)!= NULL) {
if (SetCpuStatInfo(line, &curCpuInfo)!= CM_SUCCESS) {
write_runlog(ERROR, "failed to set cpu stat info, line=[%s].\n", line);
FCLOSE_AND_RESET(fp);
return CM_ERROR;
}
break;
}
fclose(fp);
if (cpuStatInfo->oldCpuInfo.cpuTotalTime != 0) {
uint64* oldValue = (uint64*) &(cpuStatInfo->oldCpuInfo);
uint64* curValue = (uint64*) &(curCpuInfo);
uint32 count = (uint32)sizeof(CpuSimpleInfo) / sizeof(uint64);
for (uint32 i = 0; i < count; ++i) {
if (*(curValue++) < *(oldValue++)) {
write_runlog(ERROR, "get cpu info wrong, curCpuTime(%lu) < oldCpuTime(%lu). and index=[%u]\n",
*(curValue--), *(oldValue--), i);
errno_t rc = memcpy_s(&(cpuStatInfo->oldCpuInfo), sizeof(CpuSimpleInfo), &(curCpuInfo),
sizeof(CpuSimpleInfo));
securec_check_errno(rc, (void) rc);
return CM_ERROR;
}
}
uint64 totalTimeInterval = curCpuInfo.cpuTotalTime - cpuStatInfo->oldCpuInfo.cpuTotalTime;
uint64 sysTimeInterval = curCpuInfo.cpuSystemTime - cpuStatInfo->oldCpuInfo.cpuSystemTime;
uint64 userTimeInterval = curCpuInfo.cpuUserTime - cpuStatInfo->oldCpuInfo.cpuUserTime;
uint64 idleTimeInterval = curCpuInfo.cpuIdleTime - cpuStatInfo->oldCpuInfo.cpuIdleTime;
cpuStatInfo->cpuUtil = (float)(totalTimeInterval - idleTimeInterval) / (float)totalTimeInterval * PERCENT;
cpuStatInfo->cpuSystemUtil = (float)sysTimeInterval / (float)totalTimeInterval * PERCENT;
cpuStatInfo->cpuUserUtil = (float)userTimeInterval / (float)totalTimeInterval * PERCENT;
}
errno_t rc = memcpy_s(&(cpuStatInfo->oldCpuInfo), sizeof(CpuSimpleInfo), &(curCpuInfo), sizeof(CpuSimpleInfo));
securec_check_errno(rc, (void) rc);
return CM_SUCCESS;
}
void CalculateDiskIoUtil(DisIoStatInfo* disIoStatInfo, uint64 curTime, uint64 ioTimeMs,
uint64 readCount, uint64 writeCount)
{
uint64 rwCount = (readCount + writeCount) - (disIoStatInfo->lastReadCount + disIoStatInfo->lastWriteCount);
rwCount = (rwCount == 0) ? 1 : rwCount;
if ((disIoStatInfo->lastCheckTime == 0) || (disIoStatInfo->lastCheckTime >= curTime)
|| (disIoStatInfo->lastIoTime >= ioTimeMs) || (rwCount < 0)) {
if (disIoStatInfo->lastCheckTime >= curTime) {
write_runlog(ERROR, "lastCheckTime(%lu) >= curTime(%lu), get disk io info failed.\n",
disIoStatInfo->lastCheckTime, curTime);
}
if (disIoStatInfo->lastIoTime > ioTimeMs) {
write_runlog(ERROR, "lastIoTime(%lu) > ioTimeMs(%lu), get disk io info failed.\n",
disIoStatInfo->lastIoTime, ioTimeMs);
}
if (rwCount < 0) {
write_runlog(ERROR, "lastReadCount(%lu) + lastWriteCount(%lu) > readCount(%lu) + writeCount(%lu),"
" get disk io info failed.\n", disIoStatInfo->lastReadCount,
disIoStatInfo->lastWriteCount, readCount, writeCount);
}
disIoStatInfo->lastCheckTime = curTime;
disIoStatInfo->lastIoTime = ioTimeMs;
disIoStatInfo->lastReadCount = readCount;
disIoStatInfo->lastWriteCount = writeCount;
disIoStatInfo->ioUtil = 0;
return;
}
uint64 ioTimeInterval = ioTimeMs - disIoStatInfo->lastIoTime;
uint64 timeInterval = curTime - disIoStatInfo->lastCheckTime;
if (ioTimeInterval > timeInterval) {
ioTimeInterval = timeInterval;
}
disIoStatInfo->ioUtil = (float)ioTimeInterval / (float)timeInterval * PERCENT;
disIoStatInfo->svctm = (float)ioTimeInterval / (float)rwCount;
disIoStatInfo->lastCheckTime = curTime;
disIoStatInfo->lastIoTime = ioTimeMs;
disIoStatInfo->lastReadCount = readCount;
disIoStatInfo->lastWriteCount = writeCount;
}
void SetOneDiskStatus(DisIoStatInfo* disIoStatInfo)
{
FILE* fp = fopen(FILE_DISKSTAT, "re");
if (fp == NULL) {
write_runlog(ERROR, "failed to open file %s.\n", FILE_DISKSTAT);
return;
}
uint32 majorDeviceNum;
uint32 minorDeviceNum;
char deviceName[MAX_DEVICE_DIR] = {0};
uint64 readCount;
uint64 readMerge;
uint64 readSector;
uint64 readTimeMs;
uint64 writeCount;
uint64 writeMerge;
uint64 writeSector;
uint64 writeTimeMs;
uint64 ioCountInProgress;
uint64 ioTimeMs;
uint64 ioWeightTimeMs;
int cnt;
char line[MAX_PATH_LEN] = {0};
uint64 curTime;
while (fgets(line, sizeof(line), fp) != NULL) {
curTime = GetMonotonicTimeMs();
cnt = sscanf_s(line, "%u %u %s %lu %lu %lu %lu %lu %lu %lu %lu %lu %lu %lu",
&majorDeviceNum, &minorDeviceNum, deviceName, MAX_DEVICE_DIR - 1, &readCount, &readMerge, &readSector,
&readTimeMs, &writeCount, &writeMerge, &writeSector, &writeTimeMs,
&ioCountInProgress, &ioTimeMs, &ioWeightTimeMs);
check_sscanf_s_result(cnt, MAX_DEVICE_STAT_INDEX);
if (cnt != MAX_DEVICE_STAT_INDEX) {
write_runlog(ERROR, "get disk info from %s failed, line %s, count %d.\n", FILE_DISKSTAT, line, cnt);
continue;
}
if (strcmp(disIoStatInfo->diskName, deviceName) == 0) {
CalculateDiskIoUtil(disIoStatInfo, curTime, ioTimeMs, readCount, writeCount);
fclose(fp);
return;
}
}
fclose(fp);
}
status_t CheckDiskStatus(DiskStatInfo* diskStatInfo)
{
if (diskStatInfo->diskCount == 0) {
write_runlog(DEBUG5, "diskCount is 0, can not check disk io.\n");
return CM_ERROR;
}
for (uint32 i = 0; i < diskStatInfo->diskCount; ++i) {
if (CM_IS_EMPTY_STR(diskStatInfo->disIoStatInfo[i].diskName)) {
write_runlog(LOG, "diskName is empty, can not check disk io.\n");
continue;
}
SetOneDiskStatus(&diskStatInfo->disIoStatInfo[i]);
}
return CM_SUCCESS;
}
void CheckDnDiskUsage()
{
if (g_diskUsageThreshold == 0) {
return;
}
for (uint32 i = 0; i < g_currentNode->datanodeCount; ++i) {
uint32 diskUsage = GetDiskUsageForPath(g_currentNode->datanode[i].datanodeLocalDataPath);
if (diskUsage > g_diskUsageThreshold) {
char deviceName[MAX_PATH_LEN] = {0};
CmGetDisk(g_currentNode->datanode[i].datanodeLocalDataPath, deviceName, MAX_PATH_LEN);
write_runlog(LOG, "name %s, disk usage is %u, threshold is %u, report alarm.\n",
deviceName, diskUsage, g_diskUsageThreshold);
ReportDiskUsageAbnormalAlarm(deviceName, diskUsage, g_diskUsageThreshold);
}
}
}
void CheckDnDiskDamageAlarm()
{
for (uint32 i = 0; i < g_currentNode->datanodeCount; ++i) {
CheckDnDiskDamage(i);
}
}
void* CheckSysStatusThreadMain(void* const arg)
{
pthread_t threadId = pthread_self();
write_runlog(LOG, "system status check thread start, threadid %lu.\n", threadId);
uint64 lastReportSysTime = GetMonotonicTimeS();
EnvThreshold threshold = {0};
SystemStatInfo systemStat;
errno_t rc = memset_s(&systemStat, sizeof(SystemStatInfo), 0, sizeof(SystemStatInfo));
securec_check_errno(rc, (void)rc);
InitSystemStatInfo(&systemStat);
for (;;) {
if (g_shutdownRequest) {
cm_sleep(SHUTDOWN_SLEEP_TIME);
break;
}
uint64 curTime = GetMonotonicTimeS();
if ((g_sys_report_interval != 0) && (curTime - lastReportSysTime >= g_sys_report_interval)) {
CheckMemoryStatus(&systemStat.memoryStatInfo);
CheckCpuStatus(&systemStat.cpuStatInfo);
CheckDiskStatus(&systemStat.diskStatInfo);
lastReportSysTime = curTime;
}
if (GetThreshold(threshold)!= CM_SUCCESS) {
threshold = {0, 0, 0, 0, 0};
}
CheckDnDiskUsage();
CheckDnDiskDamageAlarm();
ReportSystemStatusAlarm(&systemStat, &threshold);
cm_sleep(1);
}
delete[] systemStat.diskStatInfo.disIoStatInfo;
write_runlog(LOG, "system status check thread exit.\n");
return NULL;
}
int CreateCheckSysStatusThread()
{
int err;
pthread_t thrId;
if ((err = pthread_create(&thrId, NULL, CheckSysStatusThreadMain, NULL)) != 0) {
write_runlog(ERROR, "Failed to create new thread: error %d.\n", err);
return err;
}
return 0;
}
static DDB_ROLE GetCurrentEtcdRole()
{
const uint32 serverLen = 2;
ServerSocket server[serverLen] = {{0}};
SetServerSocketWithEtcdInfo(&server[0], g_currentNode);
server[1].host = NULL;
DdbInitConfig config = {DB_ETCD};
GetDdbCfgApi(&config.drvApiInfo, server, serverLen);
DdbNodeState nodeState;
DdbConn dbCon = {0};
status_t st = InitDdbConn(&dbCon, &config);
if (st != CM_SUCCESS) {
write_runlog(ERROR, "etcd open failed when query etcd status. %s\n", DdbGetLastError(&dbCon));
return DDB_ROLE_UNKNOWN;
}
st = DdbInstanceState(&dbCon, g_currentNode->etcdName, &nodeState);
if (DdbFreeConn(&dbCon) != CM_SUCCESS) {
write_runlog(WARNING, "etcd_close failed,%s\n", DdbGetLastError(&dbCon));
}
if (st != CM_SUCCESS) {
write_runlog(ERROR, "[GetCurrentEtcdRole] failed ,error is %s\n", DdbGetLastError(&dbCon));
return DDB_ROLE_UNKNOWN;
}
return nodeState.role;
}
static void StopCurrentETCD(void)
{
char command[MAXPGPATH];
int ret = snprintf_s(command, MAXPGPATH, MAXPGPATH - 1, "echo -e %s > %s; chmod 600 %s",
CM_AGENT_BIN_NAME, g_cmEtcdManualStartPath, g_cmEtcdManualStartPath);
securec_check_intval(ret, (void)ret);
ret = system(command);
if (ret != 0) {
write_runlog(ERROR, "Failed to stop the etcd node with executing the command: command=\"%s\","
" nodeId=%u, systemReturn=%d, shellReturn=%d, errno=%d.\n",
command, g_currentNode->node, ret, SHELL_RETURN_CODE(ret), errno);
}
}
static void StartCurrentETCD(void)
{
char command[MAXPGPATH];
int ret = snprintf_s(command, MAXPGPATH, MAXPGPATH - 1, "rm -f %s", g_cmEtcdManualStartPath);
securec_check_intval(ret, (void)ret);
ret = system(command);
if (ret != 0) {
write_runlog(ERROR, "Failed to start the etcd node with executing the command: command=\"%s\","
" nodeId=%u, systemReturn=%d, shellReturn=%d, errno=%d.\n",
command, g_currentNode->node, ret, SHELL_RETURN_CODE(ret), errno);
}
}
static bool IsEtcdStopByCmAgent(const char* path)
{
char stopType[MAX_PATH_LEN] = {0};
char realPath[PATH_MAX] = {0};
if (realpath(path, realPath) == NULL) {
write_runlog(DEBUG1, "Canonical etcd_manual_start file failed errno=%d.\n", errno);
return false;
}
FILE* fd = fopen(realPath, "re");
if (fd == NULL) {
write_runlog(ERROR, "Open etcd_manual_start failed \n");
return false;
}
if (fscanf_s(fd, "%s", stopType, sizeof(stopType)) != 1) {
write_runlog(ERROR, "invalid data in etcd_manual_start file \"%s\"\n", path);
(void)fclose(fd);
return false;
}
(void)fclose(fd);
if (strcmp(stopType, CM_AGENT_BIN_NAME) == 0) {
return true;
}
return false;
}
void* ETCDConnectionStatusCheckMain(void* arg)
{
pthread_t threadId = pthread_self();
write_runlog(LOG, "etcd connection status check thread start, threadid %lu.\n", threadId);
int count;
DDB_ROLE etcdRole = DDB_ROLE_UNKNOWN;
for (;;) {
set_thread_state(threadId);
if (g_shutdownRequest || g_exitFlag) {
cm_sleep(5);
continue;
}
count = 0;
for (uint32 i = 0; i < g_node_num; ++i) {
if (!g_node[i].etcd || g_currentNode->etcdId == g_node[i].etcdId) {
continue;
}
PingPeerIP(&count, g_currentNode->etcdClientListenIPs[0], g_node[i].etcdClientListenIPs[0]);
}
DDB_ROLE tmpRole = GetCurrentEtcdRole();
etcdRole = tmpRole == DDB_ROLE_UNKNOWN ? etcdRole : tmpRole;
if (count == 0) {
write_runlog(WARNING, "current etcd is disconnected from other, etcd num=%u.\n", g_etcd_num);
if (access(g_cmEtcdManualStartPath, F_OK) != 0 && etcdRole == DDB_ROLE_LEADER) {
write_runlog(WARNING, "current etcd is leader, cmagent need stop it for etcd availability.\n");
StopCurrentETCD();
}
} else {
if (IsEtcdStopByCmAgent(g_cmEtcdManualStartPath)) {
write_runlog(LOG, "current etcd is stop by cmagent and connection is normal, need start it.\n");
StartCurrentETCD();
}
}
cm_sleep(agent_report_interval);
}
return NULL;
}
#ifdef ENABLE_XALARMD
static void TrimTrailingBlanks(char *str)
{
if (str == NULL) {
return;
}
int32 len = (int32)strlen(str);
while (len > 0 && isspace((unsigned char)str[len - 1])) {
str[len - 1] = '\0';
--len;
}
}
static char *SkipLeadingBlanks(char *str)
{
if (str == NULL) {
return NULL;
}
while (*str != '\0' && isspace((unsigned char)*str)) {
++str;
}
return str;
}
static bool ParseNodeMapEntry(char *entry, XalarmNodeMapItem *mapItem)
{
char *entryStart = SkipLeadingBlanks(entry);
if (entryStart == NULL || *entryStart == '\0') {
return false;
}
TrimTrailingBlanks(entryStart);
char *firstSep = strchr(entryStart, ':');
if (firstSep == NULL) {
return false;
}
*firstSep = '\0';
char *nodeIdText = SkipLeadingBlanks(entryStart);
char *cnaText = SkipLeadingBlanks(firstSep + 1);
TrimTrailingBlanks(nodeIdText);
TrimTrailingBlanks(cnaText);
if (nodeIdText == NULL || cnaText == NULL || nodeIdText[0] == '\0' || cnaText[0] == '\0') {
return false;
}
mapItem->nodeId = (uint32)strtoul(nodeIdText, NULL, DECIMAL_BASE);
mapItem->cna = (uint32)strtoul(cnaText, NULL, DECIMAL_BASE);
return (mapItem->nodeId != 0 && mapItem->cna != 0);
}
static bool LoadXalarmNodeMap()
{
g_xalarmNodeMapCount = 0;
char mapText[XALARM_EVENT_MAP_TEXT_LEN] = {0};
if (get_config_param(configDir, "xalarm_node_map", mapText, sizeof(mapText)) < 0) {
write_runlog(LOG, "xalarm_node_map is not configured, xalarm event will not be mapped.\n");
return false;
}
char *savePtr = NULL;
char *entry = strtok_r(mapText, ";", &savePtr);
while (entry != NULL && g_xalarmNodeMapCount < XALARM_EVENT_NODE_MAP_MAX) {
XalarmNodeMapItem mapItem = {0};
if (ParseNodeMapEntry(entry, &mapItem)) {
g_xalarmNodeMap[g_xalarmNodeMapCount++] = mapItem;
} else {
write_runlog(ERROR, "invalid xalarm_node_map entry: %s.\n", entry);
}
entry = strtok_r(NULL, ";", &savePtr);
}
write_runlog(LOG, "loaded %u xalarm map items from cm_agent.conf.\n", g_xalarmNodeMapCount);
return (g_xalarmNodeMapCount > 0);
}
static uint32 FindAlarmNodeByCna(uint32 cna)
{
for (uint32 i = 0; i < g_xalarmNodeMapCount; ++i) {
if (g_xalarmNodeMap[i].cna == cna) {
return g_xalarmNodeMap[i].nodeId;
}
}
return INVALID_ALARM_NODE_ID;
}
static bool ParseRemoteNodeAlarmDesc(const char *alarmDesc, uint32 *cna)
{
const char *payload = strstr(alarmDesc, "{cna:");
if (payload == NULL) {
return false;
}
int ret = sscanf_s(payload, "{cna:%u", cna);
return (ret == 1);
}
static bool IsNeedProcessXalarmEvent(int alarmId)
{
return (alarmId == (int)XALARM_PANIC_EVENT_ID || alarmId == (int)XALARM_KERNEL_REBOOT_EVENT_ID);
}
static uint32 XalarmMapAlarmIdToAckEventId(int alarmId)
{
switch (alarmId) {
case (int)XALARM_REBOOT_EVENT_ID:
return XALARM_REBOOT_ACK_EVENT_ID;
case (int)XALARM_OOM_EVENT_ID:
return XALARM_OOM_ACK_EVENT_ID;
case (int)XALARM_PANIC_EVENT_ID:
return XALARM_PANIC_ACK_EVENT_ID;
case (int)XALARM_KERNEL_REBOOT_EVENT_ID:
return XALARM_KERNEL_REBOOT_ACK_EVENT_ID;
default:
return 0;
}
}
static void ReportXalarmEventAck(int alarmId, const char *alarmDesc)
{
uint32 ackId = XalarmMapAlarmIdToAckEventId(alarmId);
if (ackId == 0) {
return;
}
char ackBuf[8192] = {0};
errno_t rc;
if (alarmDesc != NULL && alarmDesc[0] != '\0') {
rc = strncpy_s(ackBuf, sizeof(ackBuf), alarmDesc, sizeof(ackBuf) - 1);
securec_check_errno(rc, (void)rc);
} else {
rc = snprintf_s(ackBuf, sizeof(ackBuf), sizeof(ackBuf) - 1, "{\"ack\":\"cm_agent\",\"alarm_id\":%d}", alarmId);
securec_check_intval(rc, (void)rc);
}
#ifdef CM_XALARM_REPORT_EVENT_HAS_LEN
size_t payloadLen = strlen(ackBuf);
int rptRet = xalarm_report_event((unsigned short)ackId, ackBuf, payloadLen);
#else
int rptRet = xalarm_report_event((unsigned short)ackId, ackBuf);
#endif
if (rptRet != 0) {
write_runlog(WARNING, "xalarm_report_event ack failed, ret=%d errno=%d ackId=%u srcAlarmId=%d.\n",
rptRet, errno, ackId, alarmId);
} else {
write_runlog(DEBUG1, "xalarm_report_event ack ok, ackId=%u srcAlarmId=%u.\n", ackId, (uint32)alarmId);
}
}
static bool WaitCmsPrimarySwitchByConnect(uint32 alarmNodeId, uint32 timeoutMs)
{
const uint32 retryIntervalMs = 500;
uint32 waitedMs = 0;
while (waitedMs < timeoutMs) {
if (g_shutdownRequest || g_exitFlag) {
return false;
}
CM_Conn *primaryConn = GetConnToCmserver(0);
if (primaryConn != NULL) {
uint32 primaryNodeId = g_serverNodeId;
CMPQfinish(primaryConn);
if (primaryNodeId != alarmNodeId) {
write_runlog(LOG,
"confirm cms primary switched by connect check, alarmNodeId=%u, currentPrimaryNodeId=%u.\n",
alarmNodeId, primaryNodeId);
return true;
}
write_runlog(DEBUG1,
"cms primary still on alarm node, alarmNodeId=%u, currentPrimaryNodeId=%u, waited=%u ms.\n",
alarmNodeId, primaryNodeId, waitedMs);
}
CmUsleep(retryIntervalMs * MICROSECONDS_PER_MILLISECOND);
waitedMs += retryIntervalMs;
}
write_runlog(WARNING, "wait cms primary switch by connect timeout, alarmNodeId=%u, waited %u ms.\n",
alarmNodeId, timeoutMs);
return false;
}
static void SendXalarmEventToLocalCms(const AgentToCmPanicRebootAlarmReport *alarmMsg)
{
if (g_currentNode == NULL || g_currentNode->cmServerLevel != 1) {
return;
}
CM_Conn *localConn = GetConnToLocalCmserver();
if (localConn == NULL) {
write_runlog(WARNING, "xalarm event cannot connect local cms, nodeId=%u, alarmId=%u.\n",
g_currentNode->node, alarmMsg->alarmId);
return;
}
AgentToCmPanicRebootAlarmReport localAlarmMsg = *alarmMsg;
localAlarmMsg.msgType = MSG_AGENT_CM_PANIC_REBOOT_ALARM;
if (SendMsgToCmsByConn(localConn, (const char *)&localAlarmMsg, (uint32)sizeof(localAlarmMsg)) != CM_SUCCESS) {
write_runlog(ERROR, "xalarm event report to local cms failed, nodeId=%u, alarmId=%u.\n",
g_currentNode->node, alarmMsg->alarmId);
} else {
write_runlog(LOG, "xalarm event reported to local cms first, nodeId=%u, alarmId=%u.\n",
g_currentNode->node, alarmMsg->alarmId);
}
CMPQfinish(localConn);
}
static bool SendXalarmEventToPrimaryCms(const AgentToCmPanicRebootAlarmReport *alarmMsg)
{
const uint32 retryCount = 10;
const uint32 retryIntervalMs = 500;
for (uint32 i = 0; i < retryCount; ++i) {
if (g_shutdownRequest || g_exitFlag) {
return false;
}
CM_Conn *primaryConn = GetConnToCmserver(0);
if (primaryConn != NULL) {
AgentToCmPanicRebootAlarmReport primaryAlarmMsg = *alarmMsg;
primaryAlarmMsg.msgType = MSG_AGENT_CM_PANIC_REBOOT_ALARM_TO_PRIMARY;
bool sendOk = (SendMsgToCmsByConn(
primaryConn, (const char *)&primaryAlarmMsg, (uint32)sizeof(primaryAlarmMsg)) == CM_SUCCESS);
uint32 primaryNodeId = g_serverNodeId;
CMPQfinish(primaryConn);
if (sendOk) {
write_runlog(LOG, "xalarm event reported to primary cms, primaryNodeId=%u, alarmId=%u.\n",
primaryNodeId, alarmMsg->alarmId);
return true;
}
write_runlog(ERROR, "xalarm event report to primary cms failed, retry=%u, alarmId=%u.\n",
i + 1, alarmMsg->alarmId);
}
CmUsleep(retryIntervalMs * MICROSECONDS_PER_MILLISECOND);
}
return false;
}
static void ProcessXalarmEventReport(struct alarm_info *param, int alarmId)
{
char *alarmDesc = xalarm_getdesc(param);
if (alarmDesc == NULL) {
write_runlog(ERROR, "xalarm desc is null, alarmId=%d.\n", alarmId);
return;
}
ReportXalarmEventAck(alarmId, alarmDesc);
uint32 cna = 0;
if (!ParseRemoteNodeAlarmDesc(alarmDesc, &cna)) {
write_runlog(ERROR, "failed to parse xalarm payload, alarmId=%d, payload=%s.\n", alarmId, alarmDesc);
return;
}
uint32 alarmNodeId = FindAlarmNodeByCna(cna);
if (alarmNodeId == INVALID_ALARM_NODE_ID) {
write_runlog(ERROR, "cannot find node mapping for xalarm, alarmId=%d, cna=%u.\n", alarmId, cna);
return;
}
AgentToCmPanicRebootAlarmReport alarmMsg = {0};
alarmMsg.msgType = MSG_AGENT_CM_PANIC_REBOOT_ALARM;
alarmMsg.sourceNodeId = g_currentNode->node;
alarmMsg.alarmNodeId = alarmNodeId;
alarmMsg.cna = cna;
alarmMsg.alarmId = (uint32)alarmId;
alarmMsg.alarmType = (uint32)xalarm_gettype(param);
alarmMsg.alarmTime = (int64)xalarm_gettime(param);
alarmMsg.eid[0] = '\0';
errno_t rc = strncpy_s(alarmMsg.alarmDesc, sizeof(alarmMsg.alarmDesc), alarmDesc, sizeof(alarmMsg.alarmDesc) - 1);
securec_check_errno(rc, (void)rc);
SendXalarmEventToLocalCms(&alarmMsg);
if (!WaitCmsPrimarySwitchByConnect(alarmNodeId, CMS_PRIMARY_SWITCH_WAIT_TIMEOUT_MS)) {
write_runlog(ERROR, "xalarm event wait cms primary switch timeout, alarmId=%d, nodeId=%u.\n",
alarmId, alarmNodeId);
return;
}
write_runlog(LOG, "send xalarm event to primary cms, alarmId=%d, nodeId=%u, cna=%u.\n",
alarmId, alarmNodeId, cna);
if (!SendXalarmEventToPrimaryCms(&alarmMsg)) {
write_runlog(ERROR, "report xalarm event to primary cms failed after retries, alarmId=%d, nodeId=%u.\n",
alarmId, alarmNodeId);
return;
}
write_runlog(LOG, "reported xalarm event to cms done, alarmId=%d, nodeId=%u, cna=%u.\n",
alarmId, alarmNodeId, cna);
}
static void HandleXalarmEvent(struct alarm_info *param)
{
if (g_shutdownRequest || g_exitFlag) {
return;
}
int alarmId = xalarm_getid(param);
if (IsNeedProcessXalarmEvent(alarmId)) {
write_runlog(LOG, "receive xalarm panic/kernel reboot event, alarmId=%d, alarmType=%d, alarmTime=%lld.\n",
alarmId, xalarm_gettype(param), (long long)xalarm_gettime(param));
ProcessXalarmEventReport(param, alarmId);
return;
}
switch (alarmId) {
case (int)XALARM_REBOOT_EVENT_ID:
case (int)XALARM_OOM_EVENT_ID:
write_runlog(DEBUG1, "receive reserved xalarm event, alarmId=%d, alarmType=%d.\n",
alarmId, xalarm_gettype(param));
ReportXalarmEventAck(alarmId, xalarm_getdesc(param));
break;
case (int)XALARM_UBUS_MEM_EVENT_ID:
case (int)XALARM_REBOOT_ACK_EVENT_ID:
case (int)XALARM_OOM_ACK_EVENT_ID:
case (int)XALARM_PANIC_ACK_EVENT_ID:
case (int)XALARM_KERNEL_REBOOT_ACK_EVENT_ID:
write_runlog(DEBUG1, "receive reserved xalarm event, alarmId=%d, alarmType=%d.\n",
alarmId, xalarm_gettype(param));
break;
default:
write_runlog(DEBUG1, "receive unsupported xalarm event, alarmId=%d.\n", alarmId);
break;
}
}
static void *XalarmEventCheckMain(void *arg)
{
(void)arg;
int regRet = -1;
struct alarm_register *reg = NULL;
write_runlog(LOG, "xalarm event check thread start.\n");
struct alarm_subscription_info idFilter = {0};
uint32 filterIdx = 0;
idFilter.id_list[filterIdx++] = (int)XALARM_PANIC_EVENT_ID;
idFilter.id_list[filterIdx++] = (int)XALARM_KERNEL_REBOOT_EVENT_ID;
idFilter.len = (int)filterIdx;
while (!g_shutdownRequest && !g_exitFlag) {
if (regRet < 0) {
regRet = xalarm_register_event(®, idFilter);
if (regRet < 0) {
write_runlog(ERROR, "xalarm_register_event (panic/reboot events) failed, ret=%d.\n", regRet);
cm_sleep(SHUTDOWN_SLEEP_TIME);
continue;
}
write_runlog(LOG, "xalarm event register success (event API).\n");
}
g_xalarmEventRegister = reg;
struct alarm_msg msg;
while (!g_shutdownRequest && !g_exitFlag) {
errno_t rc = memset_s(&msg, sizeof(msg), 0, sizeof(msg));
securec_check_errno(rc, (void)rc);
int getRet = xalarm_get_event(&msg, reg);
if (getRet < 0) {
write_runlog(WARNING, "xalarm_get_event (panic/reboot) failed, ret=%d errno=%d.\n", getRet, errno);
break;
}
struct alarm_info info;
CmaXalarmMsgToAlarmInfo(&msg, &info);
HandleXalarmEvent(&info);
}
if (g_xalarmEventRegister == reg && reg != NULL) {
CmaXalarmUnregisterEvent(®);
g_xalarmEventRegister = NULL;
}
if (g_shutdownRequest || g_exitFlag) {
break;
}
cm_sleep(1);
}
write_runlog(LOG, "xalarm event check thread exit.\n");
return NULL;
}
void CreateXalarmEventCheckThread(void)
{
char enableParam[BOOL_STR_MAX_LEN] = {0};
if (get_config_param(configDir, "enable_xalarm_event_check", enableParam, sizeof(enableParam)) < 0) {
if (get_config_param(configDir, "enable_xalarm_panic_reboot_check", enableParam, sizeof(enableParam)) < 0) {
write_runlog(LOG, "enable_xalarm_event_check is not configured, xalarm event thread is skipped.\n");
return;
}
}
if (!CheckBoolConfigParam(enableParam)) {
write_runlog(ERROR, "invalid enable_xalarm_event_check value: %s,"
"xalarm event thread is skipped.\n", enableParam);
return;
}
if (!IsBoolCmParamTrue(enableParam)) {
write_runlog(LOG, "xalarm event check disabled by config.\n");
return;
}
if (!LoadXalarmNodeMap()) {
write_runlog(ERROR, "xalarm event check enabled, but xalarm_node_map is invalid.\n");
return;
}
pthread_t threadId;
if (pthread_create(&threadId, NULL, XalarmEventCheckMain, NULL) != 0) {
write_runlog(ERROR, "failed to create xalarm event check thread.\n");
return;
}
}
static AlarmType GetAlarmType(int alarmtype)
{
if (alarmtype == XALARM_TYPE_OCCUR) {
return ALM_AT_Fault;
} else if (alarmtype == XALARM_TYPE_RECOVER) {
return ALM_AT_Resume;
}
write_runlog(ERROR, "Unknown alarm type: %d\n", alarmtype);
return ALM_AT_Fault;
}
static void BuildAlarmDetails(char *details, size_t maxSize, AlarmType alarmType,
const char *driverName, const char *dataPath)
{
errno_t rc;
if (alarmType == ALM_AT_Fault) {
rc = snprintf_s(details, maxSize, maxSize - 1,
"detected slow disk by xalarm plugin, data path: %s",
dataPath);
securec_check_intval(rc, (void)rc);
} else {
write_runlog(ERROR, "receive recover alarm type: %d\n", alarmType);
rc = snprintf_s(details, maxSize, maxSize - 1,
"recovered by xalarm plugin, data path: %s",
dataPath);
securec_check_intval(rc, (void)rc);
}
}
static void HandlePrimarySlowDisk(const char *dataPath, int index)
{
write_runlog(LOG, "Checking datanode role: index=%d, role=%d, path=%s\n",
index, g_dnReportMsg[index].dnStatus.reportMsg.local_status.local_role, dataPath);
if (g_dnReportMsg[index].dnStatus.reportMsg.local_status.local_role == INSTANCE_ROLE_PRIMARY) {
immediate_stop_one_instance(dataPath, INSTANCE_DN);
} else {
write_runlog(LOG, "Slow disk detected on non-primary node, no restart needed: role=%d, path=%s\n",
g_dnReportMsg[index].dnStatus.reportMsg.local_status.local_role, dataPath);
}
}
static bool IsDeviceNameMatching(const char* deviceName, const char* driverName)
{
if (strlen(deviceName) <= 0 || strlen(driverName) <= 0) {
return false;
}
if (strncmp(deviceName, driverName, strlen(driverName)) != 0) {
return false;
}
size_t driverNameLen = strlen(driverName);
char nextChar = deviceName[driverNameLen];
return (nextChar == '\0' || (nextChar == 'p' && isdigit(deviceName[driverNameLen + 1])));
}
static bool IsValidAlarmType(int alarmtype)
{
return (alarmtype == XALARM_TYPE_OCCUR || alarmtype == XALARM_TYPE_RECOVER);
}
static void ProcessMatchedDevice(const char *driverName, int alarmtype, int index, const char *dataPath)
{
if (!IsValidAlarmType(alarmtype)) {
return;
}
AlarmType alarmType = GetAlarmType(alarmtype);
char details[MAX_PATH_LEN] = {0};
BuildAlarmDetails(details, MAX_PATH_LEN, alarmType, driverName, dataPath);
if (alarmType == ALM_AT_Fault) {
HandlePrimarySlowDisk(dataPath, index);
} else {
write_runlog(LOG, "Slow disk recovered: %s, data path: %s\n", driverName, dataPath);
}
ReportSlowDiskAlarm(driverName, alarmType, index, details);
}
void HandleXalarm(struct alarm_info *param)
{
if (g_shutdownRequest || g_exitFlag) {
return;
}
int alarmid = xalarm_getid(param);
int alarmlevel = xalarm_getlevel(param);
int alarmtype = xalarm_gettype(param);
long long alarmtime = xalarm_gettime(param);
char *alarmStr = xalarm_getdesc(param);
write_runlog(LOG, "received xalarm notification: [alarmid:%d] "
"[alarmlevel:%d] [alarmtype:%d] [alarmtime:%lld ms]\n", alarmid, alarmlevel, alarmtype, alarmtime);
cJSON *alarm_info = cJSON_Parse(alarmStr);
if (alarm_info == NULL) {
write_runlog(ERROR, "cannot parse alarm info from xalarm.\n");
return;
}
if (!cJSON_IsObject(alarm_info)) {
write_runlog(ERROR, "alarm_info not found or not an object\n");
cJSON_Delete(alarm_info);
return;
}
cJSON *driver_name = cJSON_GetObjectItem(alarm_info, "driver_name");
if (!cJSON_IsString(driver_name) || (driver_name->valuestring == NULL)) {
write_runlog(ERROR, "driver_name not found or not a string.\n");
cJSON_Delete(alarm_info);
return;
}
for (int i = 0; i < (int)g_currentNode->datanodeCount; ++i) {
char deviceName[MAX_PATH_LEN] = {0};
const char *dataPath = g_currentNode->datanode[i].datanodeLocalDataPath;
CmGetDisk(dataPath, deviceName, MAX_PATH_LEN);
write_runlog(LOG, "CmGetDisk: path=%s, deviceName=%s, driver_name=%s， datanodeCount=%d\n",
dataPath, deviceName, driver_name->valuestring, (int)g_currentNode->datanodeCount);
if (IsDeviceNameMatching(deviceName, driver_name->valuestring)) {
ProcessMatchedDevice(driver_name->valuestring, alarmtype, i, dataPath);
break;
}
}
cJSON_Delete(alarm_info);
}
#endif
