#4829

Committed 30 Oct 2025 09:25AM UTC coverage: 49.734% (-11.3%) from 61.071%

Build # #4829

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

Merge pull request #33435 from taosdata/3.0

merge 3.0

Run Details

123072 of 323930 branches covered (37.99%)

Branch coverage included in aggregate %.

7 of 25 new or added lines in 3 files covered. (28.0%)

35232 existing lines in 327 files now uncovered.

172062 of 269495 relevant lines covered (63.85%)

70709785.06 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

53.06

/source/dnode/mgmt/node_mgmt/src/dmEnv.c

/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

#define _DEFAULT_SOURCE
// clang-format off
//#include "storageapi.h"
#include "dmMgmt.h"
#include "audit.h"
#include "libs/function/tudf.h"
#include "metrics.h"
#include "tgrant.h"
#include "tcompare.h"
#include "tss.h"
#include "tanalytics.h"
#include "stream.h"
// clang-format on

#define DM_INIT_AUDIT()                       \
  do {                                        \
    auditCfg.port = tsMonitorPort;            \
    auditCfg.server = tsMonitorFqdn;          \
    auditCfg.comp = tsMonitorComp;            \
    if ((code = auditInit(&auditCfg)) != 0) { \
      return code;                            \
    }                                         \
  } while (0)

static SDnode globalDnode = {0};

SDnode *dmInstance() { return &globalDnode; }

static int32_t dmCheckRepeatInit(SDnode *pDnode) {
  int32_t code = 0;
  if (atomic_val_compare_exchange_8(&pDnode->once, DND_ENV_INIT, DND_ENV_READY) != DND_ENV_INIT) {
    dError("env is already initialized");
    code = TSDB_CODE_REPEAT_INIT;
    return code;
  }
  return 0;
}

static int32_t dmInitSystem() {
  if (taosIgnSIGPIPE() != 0) {
    dError("failed to ignore SIGPIPE");
  }

  if (taosBlockSIGPIPE() != 0) {
    dError("failed to block SIGPIPE");
  }

  taosResolveCRC();
  return 0;
}

static int32_t dmInitMonitor() {
  int32_t code = 0;
  SMonCfg monCfg = {0};

  monCfg.maxLogs = tsMonitorMaxLogs;
  monCfg.port = tsMonitorPort;
  monCfg.server = tsMonitorFqdn;
  monCfg.comp = tsMonitorComp;
  if ((code = monInit(&monCfg)) != 0) {
    dError("failed to init monitor since %s", tstrerror(code));
  }
  return code;
}

static int32_t dmInitMetrics() {
  int32_t code = 0;
  if ((code = initMetricsManager()) != 0) {
    dError("failed to init metrics since %s", tstrerror(code));
  }
  return code;
}

static int32_t dmInitAudit() {
  SAuditCfg auditCfg = {0};
  int32_t   code = 0;

  DM_INIT_AUDIT();

  return 0;
}

static bool dmDataSpaceAvailable() {
  SDnode *pDnode = dmInstance();
  if (pDnode->pTfs) {
    return tfsDiskSpaceAvailable(pDnode->pTfs, 0);
  }
  if (!osDataSpaceAvailable()) {
    dError("data disk space unavailable, i.e. %s", tsDataDir);
    return false;
  }
  return true;
}

static int32_t dmCheckDiskSpace() {
  // availability
  int32_t code = 0;
  code = osUpdate();
  if (code != 0) {
    dError("failed to update os info since %s", tstrerror(code));
    code = 0;  // ignore the error, just log it
  }
  if (!dmDataSpaceAvailable()) {
    code = TSDB_CODE_NO_DISKSPACE;
    return code;
  }
  if (!osLogSpaceAvailable()) {
    dError("log disk space unavailable, i.e. %s", tsLogDir);
    code = TSDB_CODE_NO_DISKSPACE;
    return code;
  }
  if (!osTempSpaceAvailable()) {
    dError("temp disk space unavailable, i.e. %s", tsTempDir);
    code = TSDB_CODE_NO_DISKSPACE;
    return code;
  }
  return code;
}

int32_t dmDiskInit() {
  SDnode  *pDnode = dmInstance();
  SDiskCfg dCfg = {.level = 0, .primary = 1, .disable = 0};
  tstrncpy(dCfg.dir, tsDataDir, TSDB_FILENAME_LEN);
  SDiskCfg *pDisks = tsDiskCfg;
  int32_t   numOfDisks = tsDiskCfgNum;
  if (numOfDisks <= 0 || pDisks == NULL) {
    pDisks = &dCfg;
    numOfDisks = 1;
  }

  int32_t code = tfsOpen(pDisks, numOfDisks, &pDnode->pTfs);
  if (code != 0) {
    dError("failed to init tfs since %s", tstrerror(code));
    TAOS_RETURN(code);
  }
  return 0;
}

int32_t dmDiskClose() {
  SDnode *pDnode = dmInstance();
  tfsClose(pDnode->pTfs);
  pDnode->pTfs = NULL;
  return 0;
}

static bool dmCheckDataDirVersion() {
  char checkDataDirJsonFileName[PATH_MAX] = {0};
  snprintf(checkDataDirJsonFileName, PATH_MAX, "%s/dnode/dnodeCfg.json", tsDataDir);
  if (taosCheckExistFile(checkDataDirJsonFileName)) {
    dError("The default data directory %s contains old data of tdengine 2.x, please clear it before running!",
           tsDataDir);
    return false;
  }
  return true;
}

static int32_t dmCheckDataDirVersionWrapper() {
  if (!dmCheckDataDirVersion()) {
    return TSDB_CODE_INVALID_DATA_FMT;
  }
  return 0;
}

int32_t dmInit() {
  dInfo("start to init dnode env");
  int32_t code = 0;

#ifdef USE_SHARED_STORAGE
  if (tsSsEnabled) {
    if ((code = tssInit()) != 0) return code;
    if ((code = tssCreateDefaultInstance()) != 0) return code;
  }
#endif

  if ((code = dmDiskInit()) != 0) return code;
  if (!dmCheckDataDirVersion()) {
    code = TSDB_CODE_INVALID_DATA_FMT;
    return code;
  }
  SDnode* pDnode = dmInstance();
  if ((code = dmCheckDiskSpace()) != 0) return code;
  if ((code = dmCheckRepeatInit(pDnode)) != 0) return code;
  if ((code = dmInitSystem()) != 0) return code;
  if ((code = dmInitMonitor()) != 0) return code;
  if ((code = dmInitMetrics()) != 0) return code;
  if ((code = dmInitAudit()) != 0) return code;
  if ((code = dmInitDnode(pDnode)) != 0) return code;
  if ((code = InitRegexCache() != 0)) return code;

  gExecInfoInit(&pDnode->data, (getDnodeId_f)dmGetDnodeId, dmGetMnodeEpSet);
  if ((code = streamInit(&pDnode->data, (getDnodeId_f)dmGetDnodeId, dmGetMnodeEpSet, dmGetSynEpset)) != 0) return code;

  dInfo("dnode env is initialized");
  return 0;
}

static int32_t dmCheckRepeatCleanup(SDnode *pDnode) {
  if (atomic_val_compare_exchange_8(&pDnode->once, DND_ENV_READY, DND_ENV_CLEANUP) != DND_ENV_READY) {
    dError("dnode env is already cleaned up");
    return -1;
  }
  return 0;
}

void dmCleanup() {
  dDebug("start to cleanup dnode env");
  SDnode *pDnode = dmInstance();
  if (dmCheckRepeatCleanup(pDnode) != 0) return;
  dmCleanupDnode(pDnode);
  monCleanup();
  auditCleanup();
  syncCleanUp();
  walCleanUp();
  cleanupMetrics();
  if (udfcClose() != 0) {
    dError("failed to close udfc");
  }
  udfStopUdfd();
  taosAnalyticsCleanup();
  taosStopCacheRefreshWorker();
  (void)dmDiskClose();
  DestroyRegexCache();

#ifdef USE_SHARED_STORAGE
  if (tsSsEnabled) {
    tssCloseDefaultInstance();
    tssUninit();
  }
#endif

  dInfo("dnode env is cleaned up");

  taosMemPoolClose(gMemPoolHandle);
  taosCleanupCfg();
  taosCloseLog();
}

void dmStop() {
  SDnode *pDnode = dmInstance();
  pDnode->stop = true;
}

int32_t dmRun() {
  SDnode *pDnode = dmInstance();
  return dmRunDnode(pDnode);
}

static int32_t dmProcessCreateNodeReq(EDndNodeType ntype, SRpcMsg *pMsg) {
  int32_t code = 0;
  SDnode *pDnode = dmInstance();

  SMgmtWrapper *pWrapper = dmAcquireWrapper(pDnode, ntype);
  if (pWrapper != NULL) {
    dmReleaseWrapper(pWrapper);
    switch (ntype) {
      case MNODE:
        code = TSDB_CODE_MNODE_ALREADY_DEPLOYED;
        break;
      case QNODE:
        code = TSDB_CODE_QNODE_ALREADY_DEPLOYED;
        break;
      case SNODE:
        code = TSDB_CODE_SNODE_ALREADY_DEPLOYED;
        break;
      case BNODE:
        code = TSDB_CODE_BNODE_ALREADY_DEPLOYED;
        break;
      default:
        code = TSDB_CODE_APP_ERROR;
    }
    dError("failed to create node since %s", tstrerror(code));
    return code;
  }

  dInfo("start to process create-node-request");

  pWrapper = &pDnode->wrappers[ntype];

  if (taosMulMkDir(pWrapper->path) != 0) {
    dmReleaseWrapper(pWrapper);
    code = terrno;
    dError("failed to create dir:%s since %s", pWrapper->path, tstrerror(code));
    return code;
  }

  dInfo("path %s created", pWrapper->path);

  (void)taosThreadMutexLock(&pDnode->mutex);
  SMgmtInputOpt input = dmBuildMgmtInputOpt(pWrapper);

  dInfo("node:%s, start to create", pWrapper->name);
  code = (*pWrapper->func.createFp)(&input, pMsg);
  if (code != 0) {
    dError("node:%s, failed to create since %s", pWrapper->name, tstrerror(code));
  } else {
    dInfo("node:%s, has been created", pWrapper->name);
    code = dmOpenNode(pWrapper);
    if (code == 0) {
      code = dmStartNode(pWrapper);
    }
    pWrapper->deployed = true;
    pWrapper->required = true;
  }

  (void)taosThreadMutexUnlock(&pDnode->mutex);
  return code;
}


static int32_t dmProcessAlterNodeReq(EDndNodeType ntype, SRpcMsg *pMsg) {
  int32_t code = 0;
  if (SNODE != ntype) {
    dError("failed to process msgType %d since node type is NOT snode", pMsg->msgType);
    return TSDB_CODE_INVALID_MSG;
  }
  
  SDnode *pDnode = dmInstance();
  SMgmtWrapper *pWrapper = dmAcquireWrapper(pDnode, ntype);

  dInfo("start to process alter-node-request");

  pWrapper = &pDnode->wrappers[ntype];

  (void)taosThreadMutexLock(&pDnode->mutex);
  SMgmtInputOpt input = dmBuildMgmtInputOpt(pWrapper);

  dInfo("node:%s, start to update", pWrapper->name);
  code = (*pWrapper->func.createFp)(&input, pMsg);
  if (code != 0) {
    dError("node:%s, failed to update since %s", pWrapper->name, tstrerror(code));
  } else {
    dInfo("node:%s, has been updated", pWrapper->name);
  }

  (void)taosThreadMutexUnlock(&pDnode->mutex);

  dmReleaseWrapper(pWrapper);
  
  return code;
}


static int32_t dmProcessAlterNodeTypeReq(EDndNodeType ntype, SRpcMsg *pMsg) {
  int32_t code = 0;
  SDnode *pDnode = dmInstance();

  SMgmtWrapper *pWrapper = dmAcquireWrapper(pDnode, ntype);
  if (pWrapper == NULL) {
    dError("fail to process alter node type since node not exist");
    return TSDB_CODE_INVALID_MSG;
  }
  dmReleaseWrapper(pWrapper);

  dInfo("node:%s, start to process alter-node-type-request", pWrapper->name);

  pWrapper = &pDnode->wrappers[ntype];

  if (pWrapper->func.nodeRoleFp != NULL) {
    ESyncRole role = (*pWrapper->func.nodeRoleFp)(pWrapper->pMgmt);
    dInfo("node:%s, checking node role:%d", pWrapper->name, role);
    if (role == TAOS_SYNC_ROLE_VOTER) {
      dError("node:%s, failed to alter node type since node already is role:%d", pWrapper->name, role);
      code = TSDB_CODE_MNODE_ALREADY_IS_VOTER;
      return code;
    }
  }

  if (pWrapper->func.isCatchUpFp != NULL) {
    dInfo("node:%s, checking node catch up", pWrapper->name);
    if ((*pWrapper->func.isCatchUpFp)(pWrapper->pMgmt) != 1) {
      code = TSDB_CODE_MNODE_NOT_CATCH_UP;
      return code;
    }
  }

  dInfo("node:%s, catched up leader, continue to process alter-node-type-request", pWrapper->name);

  (void)taosThreadMutexLock(&pDnode->mutex);

  dInfo("node:%s, stopping node", pWrapper->name);
  dmStopNode(pWrapper);
  dInfo("node:%s, closing node", pWrapper->name);
  dmCloseNode(pWrapper);

  pWrapper = &pDnode->wrappers[ntype];
  if (taosMkDir(pWrapper->path) != 0) {
    (void)taosThreadMutexUnlock(&pDnode->mutex);
    code = terrno;
    dError("failed to create dir:%s since %s", pWrapper->path, tstrerror(code));
    return code;
  }

  SMgmtInputOpt input = dmBuildMgmtInputOpt(pWrapper);

  dInfo("node:%s, start to create", pWrapper->name);
  code = (*pWrapper->func.createFp)(&input, pMsg);
  if (code != 0) {
    dError("node:%s, failed to create since %s", pWrapper->name, tstrerror(code));
  } else {
    dInfo("node:%s, has been created", pWrapper->name);
    code = dmOpenNode(pWrapper);
    if (code == 0) {
      code = dmStartNode(pWrapper);
    }
    pWrapper->deployed = true;
    pWrapper->required = true;
  }

  (void)taosThreadMutexUnlock(&pDnode->mutex);
  return code;
}

static int32_t dmProcessDropNodeReq(EDndNodeType ntype, SRpcMsg *pMsg) {
  int32_t code = 0;
  SDnode *pDnode = dmInstance();

  SMgmtWrapper *pWrapper = dmAcquireWrapper(pDnode, ntype);
  if (pWrapper == NULL) {
    switch (ntype) {
      case MNODE:
        code = TSDB_CODE_MNODE_NOT_DEPLOYED;
        break;
      case QNODE:
        code = TSDB_CODE_QNODE_NOT_DEPLOYED;
        break;
      case SNODE:
        code = TSDB_CODE_SNODE_NOT_DEPLOYED;
        break;
      case BNODE:
        code = TSDB_CODE_BNODE_NOT_DEPLOYED;
        break;
      default:
        code = TSDB_CODE_APP_ERROR;
    }

    dError("failed to drop node since %s", tstrerror(code));
    return terrno = code;
  }

  (void)taosThreadMutexLock(&pDnode->mutex);
  SMgmtInputOpt input = dmBuildMgmtInputOpt(pWrapper);

  dInfo("node:%s, start to drop", pWrapper->name);
  code = (*pWrapper->func.dropFp)(&input, pMsg);
  if (code != 0) {
    dError("node:%s, failed to drop since %s", pWrapper->name, tstrerror(code));
  } else {
    dInfo("node:%s, has been dropped", pWrapper->name);
    pWrapper->required = false;
    pWrapper->deployed = false;
  }

  dmReleaseWrapper(pWrapper);

  if (code == 0) {
    dmStopNode(pWrapper);
    dmCloseNode(pWrapper);
    taosRemoveDir(pWrapper->path);
  }
  (void)taosThreadMutexUnlock(&pDnode->mutex);
  return code;
}

SMgmtInputOpt dmBuildMgmtInputOpt(SMgmtWrapper *pWrapper) {
  SMgmtInputOpt opt = {
      .path = pWrapper->path,
      .name = pWrapper->name,
      .pTfs = pWrapper->pDnode->pTfs,
      .pData = &pWrapper->pDnode->data,
      .processCreateNodeFp = dmProcessCreateNodeReq,
      .processAlterNodeFp = dmProcessAlterNodeReq,
      .processAlterNodeTypeFp = dmProcessAlterNodeTypeReq,
      .processDropNodeFp = dmProcessDropNodeReq,
      .sendMonitorReportFp = dmSendMonitorReport,
      .sendMetricsReportFp = dmSendMetricsReport,
      .monitorCleanExpiredSamplesFp = dmMonitorCleanExpiredSamples,
      .metricsCleanExpiredSamplesFp = dmMetricsCleanExpiredSamples,
      .sendAuditRecordFp = auditSendRecordsInBatch,
      .getVnodeLoadsFp = dmGetVnodeLoads,
      .getVnodeLoadsLiteFp = dmGetVnodeLoadsLite,
      .setVnodeSyncTimeoutFp = dmSetVnodeSyncTimeout,
      .getMnodeLoadsFp = dmGetMnodeLoads,
      .setMnodeSyncTimeoutFp = dmSetMnodeSyncTimeout,
      .getQnodeLoadsFp = dmGetQnodeLoads,
      .stopDnodeFp = dmStop,
  };

  opt.msgCb = dmGetMsgcb(pWrapper->pDnode);
  return opt;
}

void dmReportStartup(const char *pName, const char *pDesc) {
  SStartupInfo *pStartup = &(dmInstance()->startup);
  tstrncpy(pStartup->name, pName, TSDB_STEP_NAME_LEN);
  tstrncpy(pStartup->desc, pDesc, TSDB_STEP_DESC_LEN);
  dDebug("step:%s, %s", pStartup->name, pStartup->desc);
}

int64_t dmGetClusterId() { return globalDnode.data.clusterId; }

bool dmReadyForTest() { return dmInstance()->data.dnodeVer > 0; }

1	/*
2	* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3	*
4	* This program is free software: you can use, redistribute, and/or modify
5	* it under the terms of the GNU Affero General Public License, version 3
6	* or later ("AGPL"), as published by the Free Software Foundation.
7	*
8	* This program is distributed in the hope that it will be useful, but WITHOUT
9	* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10	* FITNESS FOR A PARTICULAR PURPOSE.
11	*
12	* You should have received a copy of the GNU Affero General Public License
13	* along with this program. If not, see <http://www.gnu.org/licenses/>.
14	*/
15
16	#define _DEFAULT_SOURCE
17	// clang-format off
18	//#include "storageapi.h"
19	#include "dmMgmt.h"
20	#include "audit.h"
21	#include "libs/function/tudf.h"
22	#include "metrics.h"
23	#include "tgrant.h"
24	#include "tcompare.h"
25	#include "tss.h"
26	#include "tanalytics.h"
27	#include "stream.h"
28	// clang-format on
29
30	#define DM_INIT_AUDIT() \
31	do { \
32	auditCfg.port = tsMonitorPort; \
33	auditCfg.server = tsMonitorFqdn; \
34	auditCfg.comp = tsMonitorComp; \
35	if ((code = auditInit(&auditCfg)) != 0) { \
36	return code; \
37	} \
38	} while (0)
39
40	static SDnode globalDnode = {0};
41
42	SDnode *dmInstance() { return &globalDnode; }	88,233,325✔
43
44	static int32_t dmCheckRepeatInit(SDnode *pDnode) {	141,230✔
45	int32_t code = 0;	141,230✔
46	if (atomic_val_compare_exchange_8(&pDnode->once, DND_ENV_INIT, DND_ENV_READY) != DND_ENV_INIT) {	141,230!
47	dError("env is already initialized");	×
48	code = TSDB_CODE_REPEAT_INIT;	×
49	return code;	×
50	}
51	return 0;	141,230✔
52	}
53
54	static int32_t dmInitSystem() {	141,230✔
55	if (taosIgnSIGPIPE() != 0) {	141,230!
56	dError("failed to ignore SIGPIPE");	×
57	}
58
59	if (taosBlockSIGPIPE() != 0) {	141,230!
60	dError("failed to block SIGPIPE");	×
61	}
62
63	taosResolveCRC();	141,230✔
64	return 0;	141,230✔
65	}
66
67	static int32_t dmInitMonitor() {	141,230✔
68	int32_t code = 0;	141,230✔
69	SMonCfg monCfg = {0};	141,230✔
70
71	monCfg.maxLogs = tsMonitorMaxLogs;	141,230✔
72	monCfg.port = tsMonitorPort;	141,230✔
73	monCfg.server = tsMonitorFqdn;	141,230✔
74	monCfg.comp = tsMonitorComp;	141,230!
75	if ((code = monInit(&monCfg)) != 0) {	141,230!
76	dError("failed to init monitor since %s", tstrerror(code));	×
77	}
78	return code;	141,230✔
79	}
80
81	static int32_t dmInitMetrics() {	141,230✔
82	int32_t code = 0;	141,230✔
83	if ((code = initMetricsManager()) != 0) {	141,230!
84	dError("failed to init metrics since %s", tstrerror(code));	×
85	}
86	return code;	141,230✔
87	}
88
89	static int32_t dmInitAudit() {	141,230✔
90	SAuditCfg auditCfg = {0};	141,230✔
91	int32_t code = 0;	141,230✔
92
93	DM_INIT_AUDIT();	141,230!
94
95	return 0;	141,230✔
96	}
97
98	static bool dmDataSpaceAvailable() {	141,230✔
99	SDnode *pDnode = dmInstance();	141,230✔
100	if (pDnode->pTfs) {	141,230!
101	return tfsDiskSpaceAvailable(pDnode->pTfs, 0);	141,230✔
102	}
103	if (!osDataSpaceAvailable()) {	×
104	dError("data disk space unavailable, i.e. %s", tsDataDir);	×
105	return false;	×
106	}
107	return true;	×
108	}
109
110	static int32_t dmCheckDiskSpace() {	141,230✔
111	// availability
112	int32_t code = 0;	141,230✔
113	code = osUpdate();	141,230✔
114	if (code != 0) {	141,230!
115	dError("failed to update os info since %s", tstrerror(code));	×
116	code = 0; // ignore the error, just log it	×
117	}
118	if (!dmDataSpaceAvailable()) {	141,230!
119	code = TSDB_CODE_NO_DISKSPACE;	×
120	return code;	×
121	}
122	if (!osLogSpaceAvailable()) {	141,230!
123	dError("log disk space unavailable, i.e. %s", tsLogDir);	×
124	code = TSDB_CODE_NO_DISKSPACE;	×
125	return code;	×
126	}
127	if (!osTempSpaceAvailable()) {	141,230!
128	dError("temp disk space unavailable, i.e. %s", tsTempDir);	×
129	code = TSDB_CODE_NO_DISKSPACE;	×
130	return code;	×
131	}
132	return code;	141,230✔
133	}
134
135	int32_t dmDiskInit() {	142,429✔
136	SDnode *pDnode = dmInstance();	142,429✔
137	SDiskCfg dCfg = {.level = 0, .primary = 1, .disable = 0};	142,429✔
138	tstrncpy(dCfg.dir, tsDataDir, TSDB_FILENAME_LEN);	142,429!
139	SDiskCfg *pDisks = tsDiskCfg;	142,429✔
140	int32_t numOfDisks = tsDiskCfgNum;	142,429✔
141	if (numOfDisks <= 0 \|\| pDisks == NULL) {	142,429!
UNCOV 142	pDisks = &dCfg;	×
UNCOV 143	numOfDisks = 1;	×
144	}
145
146	int32_t code = tfsOpen(pDisks, numOfDisks, &pDnode->pTfs);	142,429✔
147	if (code != 0) {	142,429✔
148	dError("failed to init tfs since %s", tstrerror(code));	1,199!
149	TAOS_RETURN(code);	1,199✔
150	}
151	return 0;	141,230✔
152	}
153
154	int32_t dmDiskClose() {	141,149✔
155	SDnode *pDnode = dmInstance();	141,149✔
156	tfsClose(pDnode->pTfs);	141,149✔
157	pDnode->pTfs = NULL;	141,149✔
158	return 0;	141,149✔
159	}
160
161	static bool dmCheckDataDirVersion() {	141,230✔
162	char checkDataDirJsonFileName[PATH_MAX] = {0};	141,230✔
163	snprintf(checkDataDirJsonFileName, PATH_MAX, "%s/dnode/dnodeCfg.json", tsDataDir);	141,230!
164	if (taosCheckExistFile(checkDataDirJsonFileName)) {	141,230!
165	dError("The default data directory %s contains old data of tdengine 2.x, please clear it before running!",	×
166	tsDataDir);
167	return false;	×
168	}
169	return true;	141,230✔
170	}
171
172	static int32_t dmCheckDataDirVersionWrapper() {	×
173	if (!dmCheckDataDirVersion()) {	×
174	return TSDB_CODE_INVALID_DATA_FMT;	×
175	}
176	return 0;	×
177	}
178
179	int32_t dmInit() {	142,429✔
180	dInfo("start to init dnode env");	142,429!
181	int32_t code = 0;	142,429✔
182
183	#ifdef USE_SHARED_STORAGE
184	if (tsSsEnabled) {	142,429!
185	if ((code = tssInit()) != 0) return code;	×
186	if ((code = tssCreateDefaultInstance()) != 0) return code;	×
187	}
188	#endif
189
190	if ((code = dmDiskInit()) != 0) return code;	142,429✔
191	if (!dmCheckDataDirVersion()) {	141,230!
192	code = TSDB_CODE_INVALID_DATA_FMT;	×
193	return code;	×
194	}
195	SDnode* pDnode = dmInstance();	141,230✔
196	if ((code = dmCheckDiskSpace()) != 0) return code;	141,230!
197	if ((code = dmCheckRepeatInit(pDnode)) != 0) return code;	141,230!
198	if ((code = dmInitSystem()) != 0) return code;	141,230!
199	if ((code = dmInitMonitor()) != 0) return code;	141,230!
200	if ((code = dmInitMetrics()) != 0) return code;	141,230!
201	if ((code = dmInitAudit()) != 0) return code;	141,230!
202	if ((code = dmInitDnode(pDnode)) != 0) return code;	141,230✔
203	if ((code = InitRegexCache() != 0)) return code;	141,149!
204
205	gExecInfoInit(&pDnode->data, (getDnodeId_f)dmGetDnodeId, dmGetMnodeEpSet);	141,149✔
206	if ((code = streamInit(&pDnode->data, (getDnodeId_f)dmGetDnodeId, dmGetMnodeEpSet, dmGetSynEpset)) != 0) return code;	141,149!
207
208	dInfo("dnode env is initialized");	141,149!
209	return 0;	141,149✔
210	}
211
212	static int32_t dmCheckRepeatCleanup(SDnode *pDnode) {	141,149✔
213	if (atomic_val_compare_exchange_8(&pDnode->once, DND_ENV_READY, DND_ENV_CLEANUP) != DND_ENV_READY) {	141,149!
UNCOV 214	dError("dnode env is already cleaned up");	×
UNCOV 215	return -1;	×
216	}
217	return 0;	141,149✔
218	}
219
220	void dmCleanup() {	141,149✔
221	dDebug("start to cleanup dnode env");	141,149✔
222	SDnode *pDnode = dmInstance();	141,149✔
223	if (dmCheckRepeatCleanup(pDnode) != 0) return;	141,149!
224	dmCleanupDnode(pDnode);	141,149✔
225	monCleanup();	141,149✔
226	auditCleanup();	141,149✔
227	syncCleanUp();	141,149✔
228	walCleanUp();	141,149✔
229	cleanupMetrics();	141,149✔
230	if (udfcClose() != 0) {	141,149!
231	dError("failed to close udfc");	×
232	}
233	udfStopUdfd();	141,149✔
234	taosAnalyticsCleanup();	141,149✔
235	taosStopCacheRefreshWorker();	141,149✔
236	(void)dmDiskClose();	141,149✔
237	DestroyRegexCache();	141,149✔
238
239	#ifdef USE_SHARED_STORAGE
240	if (tsSsEnabled) {	141,149!
241	tssCloseDefaultInstance();	×
242	tssUninit();	×
243	}
244	#endif
245
246	dInfo("dnode env is cleaned up");	141,149!
247
248	taosMemPoolClose(gMemPoolHandle);	141,149✔
249	taosCleanupCfg();	141,149✔
250	taosCloseLog();	141,149✔
251	}
252
253	void dmStop() {	145,430✔
254	SDnode *pDnode = dmInstance();	145,430✔
255	pDnode->stop = true;	145,430✔
256	}	145,430✔
257
258	int32_t dmRun() {	141,149✔
259	SDnode *pDnode = dmInstance();	141,149✔
260	return dmRunDnode(pDnode);	141,149✔
261	}
262
263	static int32_t dmProcessCreateNodeReq(EDndNodeType ntype, SRpcMsg *pMsg) {	19,235✔
264	int32_t code = 0;	19,235✔
265	SDnode *pDnode = dmInstance();	19,235✔
266
267	SMgmtWrapper *pWrapper = dmAcquireWrapper(pDnode, ntype);	19,235✔
268	if (pWrapper != NULL) {	19,235!
UNCOV 269	dmReleaseWrapper(pWrapper);	×
UNCOV 270	switch (ntype) {	×
271	case MNODE:	×
272	code = TSDB_CODE_MNODE_ALREADY_DEPLOYED;	×
273	break;	×
UNCOV 274	case QNODE:	×
UNCOV 275	code = TSDB_CODE_QNODE_ALREADY_DEPLOYED;	×
UNCOV 276	break;	×
277	case SNODE:	×
278	code = TSDB_CODE_SNODE_ALREADY_DEPLOYED;	×
279	break;	×
280	case BNODE:	×
281	code = TSDB_CODE_BNODE_ALREADY_DEPLOYED;	×
282	break;	×
283	default:	×
284	code = TSDB_CODE_APP_ERROR;	×
285	}
UNCOV 286	dError("failed to create node since %s", tstrerror(code));	×
UNCOV 287	return code;	×
288	}
289
290	dInfo("start to process create-node-request");	19,235!
291
292	pWrapper = &pDnode->wrappers[ntype];	19,235✔
293
294	if (taosMulMkDir(pWrapper->path) != 0) {	19,235!
295	dmReleaseWrapper(pWrapper);	×
296	code = terrno;	×
297	dError("failed to create dir:%s since %s", pWrapper->path, tstrerror(code));	×
298	return code;	×
299	}
300
301	dInfo("path %s created", pWrapper->path);	19,235!
302
303	(void)taosThreadMutexLock(&pDnode->mutex);	19,235✔
304	SMgmtInputOpt input = dmBuildMgmtInputOpt(pWrapper);	19,235✔
305
306	dInfo("node:%s, start to create", pWrapper->name);	19,235!
307	code = (*pWrapper->func.createFp)(&input, pMsg);	19,235✔
308	if (code != 0) {	19,235!
UNCOV 309	dError("node:%s, failed to create since %s", pWrapper->name, tstrerror(code));	×
310	} else {
311	dInfo("node:%s, has been created", pWrapper->name);	19,235!
312	code = dmOpenNode(pWrapper);	19,235✔
313	if (code == 0) {	19,235!
314	code = dmStartNode(pWrapper);	19,235✔
315	}
316	pWrapper->deployed = true;	19,235✔
317	pWrapper->required = true;	19,235✔
318	}
319
320	(void)taosThreadMutexUnlock(&pDnode->mutex);	19,235✔
321	return code;	19,235✔
322	}
323
324
UNCOV 325	static int32_t dmProcessAlterNodeReq(EDndNodeType ntype, SRpcMsg *pMsg) {	×
UNCOV 326	int32_t code = 0;	×
UNCOV 327	if (SNODE != ntype) {	×
328	dError("failed to process msgType %d since node type is NOT snode", pMsg->msgType);	×
329	return TSDB_CODE_INVALID_MSG;	×
330	}
331
UNCOV 332	SDnode *pDnode = dmInstance();	×
UNCOV 333	SMgmtWrapper *pWrapper = dmAcquireWrapper(pDnode, ntype);	×
334
UNCOV 335	dInfo("start to process alter-node-request");	×
336
UNCOV 337	pWrapper = &pDnode->wrappers[ntype];	×
338
UNCOV 339	(void)taosThreadMutexLock(&pDnode->mutex);	×
UNCOV 340	SMgmtInputOpt input = dmBuildMgmtInputOpt(pWrapper);	×
341
UNCOV 342	dInfo("node:%s, start to update", pWrapper->name);	×
UNCOV 343	code = (*pWrapper->func.createFp)(&input, pMsg);	×
UNCOV 344	if (code != 0) {	×
345	dError("node:%s, failed to update since %s", pWrapper->name, tstrerror(code));	×
346	} else {
UNCOV 347	dInfo("node:%s, has been updated", pWrapper->name);	×
348	}
349
UNCOV 350	(void)taosThreadMutexUnlock(&pDnode->mutex);	×
351
UNCOV 352	dmReleaseWrapper(pWrapper);	×
353
UNCOV 354	return code;	×
355	}
356
357
358	static int32_t dmProcessAlterNodeTypeReq(EDndNodeType ntype, SRpcMsg *pMsg) {	113,192✔
359	int32_t code = 0;	113,192✔
360	SDnode *pDnode = dmInstance();	113,192✔
361
362	SMgmtWrapper *pWrapper = dmAcquireWrapper(pDnode, ntype);	113,192✔
363	if (pWrapper == NULL) {	113,192!
364	dError("fail to process alter node type since node not exist");	×
365	return TSDB_CODE_INVALID_MSG;	×
366	}
367	dmReleaseWrapper(pWrapper);	113,192✔
368
369	dInfo("node:%s, start to process alter-node-type-request", pWrapper->name);	113,192!
370
371	pWrapper = &pDnode->wrappers[ntype];	113,192✔
372
373	if (pWrapper->func.nodeRoleFp != NULL) {	113,192!
374	ESyncRole role = (*pWrapper->func.nodeRoleFp)(pWrapper->pMgmt);	113,192✔
375	dInfo("node:%s, checking node role:%d", pWrapper->name, role);	113,192!
376	if (role == TAOS_SYNC_ROLE_VOTER) {	113,192!
377	dError("node:%s, failed to alter node type since node already is role:%d", pWrapper->name, role);	×
378	code = TSDB_CODE_MNODE_ALREADY_IS_VOTER;	×
379	return code;	×
380	}
381	}
382
383	if (pWrapper->func.isCatchUpFp != NULL) {	113,192!
384	dInfo("node:%s, checking node catch up", pWrapper->name);	113,192!
385	if ((*pWrapper->func.isCatchUpFp)(pWrapper->pMgmt) != 1) {	113,192✔
386	code = TSDB_CODE_MNODE_NOT_CATCH_UP;	106,040✔
387	return code;	106,040✔
388	}
389	}
390
391	dInfo("node:%s, catched up leader, continue to process alter-node-type-request", pWrapper->name);	7,152!
392
393	(void)taosThreadMutexLock(&pDnode->mutex);	7,152✔
394
395	dInfo("node:%s, stopping node", pWrapper->name);	7,152!
396	dmStopNode(pWrapper);	7,152✔
397	dInfo("node:%s, closing node", pWrapper->name);	7,152!
398	dmCloseNode(pWrapper);	7,152✔
399
400	pWrapper = &pDnode->wrappers[ntype];	7,152✔
401	if (taosMkDir(pWrapper->path) != 0) {	7,152!
402	(void)taosThreadMutexUnlock(&pDnode->mutex);	×
403	code = terrno;	×
404	dError("failed to create dir:%s since %s", pWrapper->path, tstrerror(code));	×
405	return code;	×
406	}
407
408	SMgmtInputOpt input = dmBuildMgmtInputOpt(pWrapper);	7,152✔
409
410	dInfo("node:%s, start to create", pWrapper->name);	7,152!
411	code = (*pWrapper->func.createFp)(&input, pMsg);	7,152✔
412	if (code != 0) {	7,152!
413	dError("node:%s, failed to create since %s", pWrapper->name, tstrerror(code));	×
414	} else {
415	dInfo("node:%s, has been created", pWrapper->name);	7,152!
416	code = dmOpenNode(pWrapper);	7,152✔
417	if (code == 0) {	7,152!
418	code = dmStartNode(pWrapper);	7,152✔
419	}
420	pWrapper->deployed = true;	7,152✔
421	pWrapper->required = true;	7,152✔
422	}
423
424	(void)taosThreadMutexUnlock(&pDnode->mutex);	7,152✔
425	return code;	7,152✔
426	}
427
428	static int32_t dmProcessDropNodeReq(EDndNodeType ntype, SRpcMsg *pMsg) {	280✔
429	int32_t code = 0;	280✔
430	SDnode *pDnode = dmInstance();	280✔
431
432	SMgmtWrapper *pWrapper = dmAcquireWrapper(pDnode, ntype);	280✔
433	if (pWrapper == NULL) {	280!
UNCOV 434	switch (ntype) {	×
435	case MNODE:	×
436	code = TSDB_CODE_MNODE_NOT_DEPLOYED;	×
437	break;	×
UNCOV 438	case QNODE:	×
UNCOV 439	code = TSDB_CODE_QNODE_NOT_DEPLOYED;	×
UNCOV 440	break;	×
441	case SNODE:	×
442	code = TSDB_CODE_SNODE_NOT_DEPLOYED;	×
443	break;	×
444	case BNODE:	×
445	code = TSDB_CODE_BNODE_NOT_DEPLOYED;	×
446	break;	×
447	default:	×
448	code = TSDB_CODE_APP_ERROR;	×
449	}
450
UNCOV 451	dError("failed to drop node since %s", tstrerror(code));	×
UNCOV 452	return terrno = code;	×
453	}
454
455	(void)taosThreadMutexLock(&pDnode->mutex);	280✔
456	SMgmtInputOpt input = dmBuildMgmtInputOpt(pWrapper);	280✔
457
458	dInfo("node:%s, start to drop", pWrapper->name);	280!
459	code = (*pWrapper->func.dropFp)(&input, pMsg);	280✔
460	if (code != 0) {	280!
461	dError("node:%s, failed to drop since %s", pWrapper->name, tstrerror(code));	×
462	} else {
463	dInfo("node:%s, has been dropped", pWrapper->name);	280!
464	pWrapper->required = false;	280✔
465	pWrapper->deployed = false;	280✔
466	}
467
468	dmReleaseWrapper(pWrapper);	280✔
469
470	if (code == 0) {	280!
471	dmStopNode(pWrapper);	280✔
472	dmCloseNode(pWrapper);	280✔
473	taosRemoveDir(pWrapper->path);	280✔
474	}
475	(void)taosThreadMutexUnlock(&pDnode->mutex);	280✔
476	return code;	280✔
477	}
478
479	SMgmtInputOpt dmBuildMgmtInputOpt(SMgmtWrapper *pWrapper) {	1,289,865✔
480	SMgmtInputOpt opt = {	6,442,665✔
481	.path = pWrapper->path,	1,289,865✔
482	.name = pWrapper->name,	1,289,865✔
483	.pTfs = pWrapper->pDnode->pTfs,	1,289,865✔
484	.pData = &pWrapper->pDnode->data,	1,289,865✔
485	.processCreateNodeFp = dmProcessCreateNodeReq,
486	.processAlterNodeFp = dmProcessAlterNodeReq,
487	.processAlterNodeTypeFp = dmProcessAlterNodeTypeReq,
488	.processDropNodeFp = dmProcessDropNodeReq,
489	.sendMonitorReportFp = dmSendMonitorReport,
490	.sendMetricsReportFp = dmSendMetricsReport,
491	.monitorCleanExpiredSamplesFp = dmMonitorCleanExpiredSamples,
492	.metricsCleanExpiredSamplesFp = dmMetricsCleanExpiredSamples,
493	.sendAuditRecordFp = auditSendRecordsInBatch,
494	.getVnodeLoadsFp = dmGetVnodeLoads,
495	.getVnodeLoadsLiteFp = dmGetVnodeLoadsLite,
496	.setVnodeSyncTimeoutFp = dmSetVnodeSyncTimeout,
497	.getMnodeLoadsFp = dmGetMnodeLoads,
498	.setMnodeSyncTimeoutFp = dmSetMnodeSyncTimeout,
499	.getQnodeLoadsFp = dmGetQnodeLoads,
500	.stopDnodeFp = dmStop,
501	};
502
503	opt.msgCb = dmGetMsgcb(pWrapper->pDnode);	1,289,865✔
504	return opt;	1,289,865✔
505	}
506
507	void dmReportStartup(const char pName, const char pDesc) {	9,589,411✔
508	SStartupInfo *pStartup = &(dmInstance()->startup);	9,589,411✔
509	tstrncpy(pStartup->name, pName, TSDB_STEP_NAME_LEN);	9,589,613!
510	tstrncpy(pStartup->desc, pDesc, TSDB_STEP_DESC_LEN);	9,589,293!
511	dDebug("step:%s, %s", pStartup->name, pStartup->desc);	9,589,293✔
512	}	9,589,293✔
513
514	int64_t dmGetClusterId() { return globalDnode.data.clusterId; }	×
515
UNCOV 516	bool dmReadyForTest() { return dmInstance()->data.dnodeVer > 0; }	×

taosdata / TDengine / #4829

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous