#4912

Committed 04 Jan 2026 09:05AM UTC coverage: 64.888% (-0.1%) from 65.028%

Build # #4912

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

merge: from main to 3.0 branch #34156

Run Details

1206 of 4524 new or added lines in 22 files covered. (26.66%)

5351 existing lines in 123 files now uncovered.

194856 of 300296 relevant lines covered (64.89%)

118198896.2 hits per line

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

69.59

/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

extern void cryptUnloadProviders();

#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 defined(TD_ENTERPRISE) && defined(LINUX)
  cryptUnloadProviders();
#endif

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

taosdata / TDengine / #4912

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