#4911

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

Build # #4911

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%)

1517 existing lines in 134 files now uncovered.

195276 of 300296 relevant lines covered (65.03%)

116931714.52 hits per line

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

71.09

/source/dnode/mgmt/node_mgmt/src/dmMgmt.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
#include "dmMgmt.h"
#include "dmNodes.h"
#include "index.h"
#include "qworker.h"
#include "tcompression.h"
#include "tconv.h"
#include "tglobal.h"
#include "tgrant.h"
#include "tconv.h"
#include "stream.h"
#if defined(TD_ENTERPRISE) && defined(TD_HAS_TAOSK)
#include "taoskInt.h"
#endif

static bool dmRequireNode(SDnode *pDnode, SMgmtWrapper *pWrapper) {
  SMgmtInputOpt input = dmBuildMgmtInputOpt(pWrapper);
  input.dnodeId = pDnode->data.dnodeId;

  bool    required = false;
  int32_t code = (*pWrapper->func.requiredFp)(&input, &required);
  if (!required) {
    dDebug("node:%s, does not require startup", pWrapper->name);
  } else {
    dDebug("node:%s, required to startup", pWrapper->name);
  }

  return required;
}

int32_t dmInitDnode(SDnode *pDnode) {
  dDebug("start to create dnode");
  int32_t code = -1;
  char    path[PATH_MAX + 100] = {0};

  if ((code = dmInitVarsWrapper(pDnode)) != 0) {
    goto _OVER;
  }

  // compress module init
  tsCompressInit(tsLossyColumns, tsFPrecision, tsDPrecision, tsMaxRange, tsCurRange, (int)tsIfAdtFse, tsCompressor);

  pDnode->wrappers[DNODE].func = dmGetMgmtFunc();
  pDnode->wrappers[MNODE].func = mmGetMgmtFunc();
  pDnode->wrappers[VNODE].func = vmGetMgmtFunc();
  pDnode->wrappers[QNODE].func = qmGetMgmtFunc();
  pDnode->wrappers[SNODE].func = smGetMgmtFunc();
  pDnode->wrappers[BNODE].func = bmGetMgmtFunc();
  pDnode->wrappers[XNODE].func = xmGetMgmtFunc();

  for (EDndNodeType ntype = DNODE; ntype < NODE_END; ++ntype) {
    SMgmtWrapper *pWrapper = &pDnode->wrappers[ntype];
    pWrapper->pDnode = pDnode;
    pWrapper->name = dmNodeName(ntype);
    pWrapper->ntype = ntype;
    (void)taosThreadRwlockInit(&pWrapper->lock, NULL);

    snprintf(path, sizeof(path), "%s%s%s", tsDataDir, TD_DIRSEP, pWrapper->name);
    pWrapper->path = taosStrdup(path);
    if (pWrapper->path == NULL) {
      code = terrno;
      goto _OVER;
    }

    pWrapper->required = dmRequireNode(pDnode, pWrapper);
  }

  code = dmCheckRunning(tsDataDir, &pDnode->lockfile);
  if (code != 0) {
    goto _OVER;
  }

  if ((code = dmInitModule(pDnode)) != 0) {
    goto _OVER;
  }

  indexInit(tsNumOfCommitThreads);

  if ((code = dmInitStatusClient(pDnode)) != 0) {
    goto _OVER;
  }
  if ((code = dmInitSyncClient(pDnode)) != 0) {
    goto _OVER;
  }

  dmReportStartup("dnode-transport", "initialized");
  dDebug("dnode is created, ptr:%p", pDnode);
  code = 0;

_OVER:
  if (code != 0 && pDnode != NULL) {
    dmClearVars(pDnode);
    pDnode = NULL;
    dError("failed to create dnode since %s", tstrerror(code));
  }

  return code;
}

void dmCleanupDnode(SDnode *pDnode) {
  if (pDnode == NULL) {
    return;
  }

  dmCleanupClient(pDnode);
  dmCleanupStatusClient(pDnode);
  dmCleanupSyncClient(pDnode);
  dmCleanupServer(pDnode);

  dmClearVars(pDnode);
  rpcCleanup();
  streamCleanup();
  indexCleanup();
  taosConvDestroy();

  // compress destroy
  tsCompressExit();

  dDebug("dnode is closed, ptr:%p", pDnode);
}

int32_t dmInitVarsWrapper(SDnode *pDnode) {
  int32_t code = dmInitVars(pDnode);
  if (code == -1) {
    return terrno;
  }
  return 0;
}
int32_t dmInitVars(SDnode *pDnode) {
  int32_t     code = 0;
  SDnodeData *pData = &pDnode->data;
  pData->dnodeId = 0;
  pData->clusterId = 0;
  pData->dnodeVer = 0;
  pData->engineVer = 0;
  pData->updateTime = 0;
  pData->rebootTime = taosGetTimestampMs();
  pData->dropped = 0;
  pData->stopped = 0;
  char *machineId = NULL;
  code = tGetMachineId(&machineId);
  if (machineId) {
    tstrncpy(pData->machineId, machineId, TSDB_MACHINE_ID_LEN + 1);
    taosMemoryFreeClear(machineId);
  } else {
#if defined(TD_ENTERPRISE) && !defined(GRANTS_CFG)
    code = TSDB_CODE_DNODE_NO_MACHINE_CODE;
    return terrno = code;
#endif
  }

  pData->dnodeHash = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);
  if (pData->dnodeHash == NULL) {
    dError("failed to init dnode hash");
    return terrno;
  }

  if ((code = dmReadEps(pData)) != 0) {
    dError("failed to read file since %s", tstrerror(code));
    return code;
  }

#if defined(TD_ENTERPRISE) || defined(TD_ASTRA_TODO)
  tsiEncryptAlgorithm = pData->encryptAlgorigthm;
  tsiEncryptScope = pData->encryptScope;

#if defined(TD_HAS_TAOSK) || defined(TD_ASTRA_TODO)
  // Load local encryption keys and initialize key version
  {
    char masterKeyFile[PATH_MAX] = {0};
    char derivedKeyFile[PATH_MAX] = {0};
    snprintf(masterKeyFile, sizeof(masterKeyFile), "%s%sdnode%sconfig%smaster.bin", tsDataDir, TD_DIRSEP, TD_DIRSEP,
             TD_DIRSEP);
    snprintf(derivedKeyFile, sizeof(derivedKeyFile), "%s%sdnode%sconfig%sderived.bin", tsDataDir, TD_DIRSEP, TD_DIRSEP,
             TD_DIRSEP);

    char    svrKey[129] = {0};
    char    dbKey[129] = {0};
    char    cfgKey[129] = {0};
    char    metaKey[129] = {0};
    char    dataKey[129] = {0};
    int32_t algorithm = 0;
    int32_t cfgAlgorithm = 0;
    int32_t metaAlgorithm = 0;
    int32_t fileVersion = 0;
    int32_t keyVersion = 0;
    int64_t createTime = 0;
    int64_t svrKeyUpdateTime = 0;
    int64_t dbKeyUpdateTime = 0;

    code = taoskLoadEncryptKeys(masterKeyFile, derivedKeyFile, svrKey, dbKey, cfgKey, metaKey, dataKey, &algorithm,
                                &cfgAlgorithm, &metaAlgorithm, &fileVersion, &keyVersion, &createTime, 
                                &svrKeyUpdateTime, &dbKeyUpdateTime);
    if (code == 0) {
      tsLocalKeyVersion = keyVersion;
      dInfo("loaded local encryption keys, version:%d", tsLocalKeyVersion);
    } else {
      tsLocalKeyVersion = 0;
      dInfo("no local encryption keys found or failed to load, will sync from mnode");
    }
  }
#endif
#endif
  /*
  if(tsiEncryptAlgorithm != 0) {
    if(pData->machineId != NULL && strlen(pData->machineId) > 0){
      dInfo("get crypt key at startup, machineId:%s", pData->machineId);
      int32_t code = 0;

      //code = taosGetCryptKey(tsAuthCode, pData->machineId, tsCryptKey);
      code = 0;
      tstrncpy(tsEncryptKey, tsAuthCode, 16);

      if (code != 0) {
        if(code == -1){
          terrno = TSDB_CODE_DNODE_NO_ENCRYPT_KEY;
          dError("machine code changed, can't get crypt key");
        }
        if(code == -2){
          terrno = TSDB_CODE_DNODE_NO_ENCRYPT_KEY;
          dError("failed to get crypt key");
        }
        return -1;
      }

      if(strlen(tsEncryptKey) == 0){
        terrno = TSDB_CODE_DNODE_NO_ENCRYPT_KEY;
        dError("failed to get crypt key at startup since key is null, machineId:%s", pData->machineId);
        return -1;
      }
    }
    else{
      terrno = TSDB_CODE_DNODE_NO_MACHINE_CODE;
      dError("failed to get crypt key at startup, machineId:%s", pData->machineId);
      return -1;
    }
  }
  */

  if (pData->dropped) {
    dError("dnode will not start since its already dropped");
    return -1;
  }

  (void)taosThreadRwlockInit(&pData->lock, NULL);
  (void)taosThreadMutexInit(&pData->statusInfolock, NULL);
  (void)taosThreadMutexInit(&pDnode->mutex, NULL);
  return 0;
}

extern SMonVloadInfo tsVinfo;

void dmClearVars(SDnode *pDnode) {
  for (EDndNodeType ntype = DNODE; ntype < NODE_END; ++ntype) {
    SMgmtWrapper *pWrapper = &pDnode->wrappers[ntype];
    taosMemoryFreeClear(pWrapper->path);
    (void)taosThreadRwlockDestroy(&pWrapper->lock);
  }
  if (pDnode->lockfile != NULL) {
    if (taosUnLockFile(pDnode->lockfile) != 0) {
      dError("failed to unlock file");
    }

    (void)taosCloseFile(&pDnode->lockfile);
    pDnode->lockfile = NULL;
  }

  SDnodeData *pData = &pDnode->data;
  (void)taosThreadRwlockWrlock(&pData->lock);
  if (pData->oldDnodeEps != NULL) {
    if (dmWriteEps(pData) == 0) {
      dmRemoveDnodePairs(pData);
    }
    taosArrayDestroy(pData->oldDnodeEps);
    pData->oldDnodeEps = NULL;
  }
  if (pData->dnodeEps != NULL) {
    taosArrayDestroy(pData->dnodeEps);
    pData->dnodeEps = NULL;
  }
  if (pData->dnodeHash != NULL) {
    taosHashCleanup(pData->dnodeHash);
    pData->dnodeHash = NULL;
  }
  (void)taosThreadRwlockUnlock(&pData->lock);

  (void)taosThreadRwlockDestroy(&pData->lock);

  dDebug("begin to lock status info when thread exit");
  if (taosThreadMutexLock(&pData->statusInfolock) != 0) {
    dError("failed to lock status info lock");
    return;
  }
  if (tsVinfo.pVloads != NULL) {
    taosArrayDestroy(tsVinfo.pVloads);
    tsVinfo.pVloads = NULL;
  }
  if (taosThreadMutexUnlock(&pData->statusInfolock) != 0) {
    dError("failed to unlock status info lock");
    return;
  }
  if (taosThreadMutexDestroy(&pData->statusInfolock) != 0) {
    dError("failed to destroy status info lock");
  }
  memset(&pData->statusInfolock, 0, sizeof(pData->statusInfolock));

  (void)taosThreadMutexDestroy(&pDnode->mutex);
  memset(&pDnode->mutex, 0, sizeof(pDnode->mutex));
}

void dmSetStatus(SDnode *pDnode, EDndRunStatus status) {
  if (pDnode->status != status) {
    dDebug("dnode status set from %s to %s", dmStatStr(pDnode->status), dmStatStr(status));
    pDnode->status = status;
  }
}

SMgmtWrapper *dmAcquireWrapper(SDnode *pDnode, EDndNodeType ntype) {
  SMgmtWrapper *pWrapper = &pDnode->wrappers[ntype];
  SMgmtWrapper *pRetWrapper = pWrapper;

  (void)taosThreadRwlockRdlock(&pWrapper->lock);
  if (pWrapper->deployed) {
    int32_t refCount = atomic_add_fetch_32(&pWrapper->refCount, 1);
    // dTrace("node:%s, is acquired, ref:%d", pWrapper->name, refCount);
  } else {
    pRetWrapper = NULL;
  }
  (void)taosThreadRwlockUnlock(&pWrapper->lock);

  return pRetWrapper;
}

int32_t dmMarkWrapper(SMgmtWrapper *pWrapper) {
  int32_t code = 0;

  (void)taosThreadRwlockRdlock(&pWrapper->lock);
  if (pWrapper->deployed) {
    int32_t refCount = atomic_add_fetch_32(&pWrapper->refCount, 1);
    // dTrace("node:%s, is marked, ref:%d", pWrapper->name, refCount);
  } else {
    switch (pWrapper->ntype) {
      case MNODE:
        code = TSDB_CODE_MNODE_NOT_FOUND;
        break;
      case QNODE:
        code = TSDB_CODE_QNODE_NOT_FOUND;
        break;
      case SNODE:
        code = TSDB_CODE_SNODE_NOT_FOUND;
        break;
      case BNODE:
        code = TSDB_CODE_BNODE_NOT_FOUND;
        break;
      case XNODE:
        code = TSDB_CODE_XNODE_NOT_FOUND;
        break;
      case VNODE:
        code = TSDB_CODE_VND_STOPPED;
        break;
      default:
        code = TSDB_CODE_APP_IS_STOPPING;
        break;
    }
  }
  (void)taosThreadRwlockUnlock(&pWrapper->lock);

  return code;
}

void dmReleaseWrapper(SMgmtWrapper *pWrapper) {
  if (pWrapper == NULL) return;

  (void)taosThreadRwlockRdlock(&pWrapper->lock);
  int32_t refCount = atomic_sub_fetch_32(&pWrapper->refCount, 1);
  (void)taosThreadRwlockUnlock(&pWrapper->lock);
  // dTrace("node:%s, is released, ref:%d", pWrapper->name, refCount);
}

static void dmGetServerStartupStatus(SDnode *pDnode, SServerStatusRsp *pStatus) {
  SDnodeMgmt *pMgmt = pDnode->wrappers[DNODE].pMgmt;
  pStatus->details[0] = 0;

  if (pDnode->status == DND_STAT_INIT) {
    pStatus->statusCode = TSDB_SRV_STATUS_NETWORK_OK;
    snprintf(pStatus->details, sizeof(pStatus->details), "%s: %s", pDnode->startup.name, pDnode->startup.desc);
  } else if (pDnode->status == DND_STAT_STOPPED) {
    pStatus->statusCode = TSDB_SRV_STATUS_EXTING;
  } else {
    pStatus->statusCode = TSDB_SRV_STATUS_SERVICE_OK;
  }
}

void dmProcessNetTestReq(SDnode *pDnode, SRpcMsg *pMsg) {
  dDebug("msg:%p, net test req will be processed", pMsg);

  SRpcMsg rsp = {.info = pMsg->info};
  rsp.pCont = rpcMallocCont(pMsg->contLen);
  if (rsp.pCont == NULL) {
    rsp.code = TSDB_CODE_OUT_OF_MEMORY;
  } else {
    rsp.contLen = pMsg->contLen;
  }

  if (rpcSendResponse(&rsp) != 0) {
    dError("failed to send response, msg:%p", &rsp);
  }
  rpcFreeCont(pMsg->pCont);
}

void dmProcessServerStartupStatus(SDnode *pDnode, SRpcMsg *pMsg) {
  dDebug("msg:%p, server startup status req will be processed", pMsg);

  SServerStatusRsp statusRsp = {0};
  dmGetServerStartupStatus(pDnode, &statusRsp);

  SRpcMsg rsp = {.info = pMsg->info};
  int32_t contLen = tSerializeSServerStatusRsp(NULL, 0, &statusRsp);
  if (contLen < 0) {
    rsp.code = TSDB_CODE_OUT_OF_MEMORY;
  } else {
    rsp.pCont = rpcMallocCont(contLen);
    if (rsp.pCont != NULL) {
      if (tSerializeSServerStatusRsp(rsp.pCont, contLen, &statusRsp) < 0) {
        rsp.code = TSDB_CODE_APP_ERROR;
      } else {
        rsp.contLen = contLen;
      }
    }
  }

  if (rpcSendResponse(&rsp) != 0) {
    dError("failed to send response, msg:%p", &rsp);
  }
  rpcFreeCont(pMsg->pCont);
}

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	#include "dmMgmt.h"
18	#include "dmNodes.h"
19	#include "index.h"
20	#include "qworker.h"
21	#include "tcompression.h"
22	#include "tconv.h"
23	#include "tglobal.h"
24	#include "tgrant.h"
25	#include "tconv.h"
26	#include "stream.h"
27	#if defined(TD_ENTERPRISE) && defined(TD_HAS_TAOSK)
28	#include "taoskInt.h"
29	#endif
30
31	static bool dmRequireNode(SDnode pDnode, SMgmtWrapper pWrapper) {	3,776,920✔
32	SMgmtInputOpt input = dmBuildMgmtInputOpt(pWrapper);	3,776,920✔
33	input.dnodeId = pDnode->data.dnodeId;	3,776,920✔
34
35	bool required = false;	3,776,920✔
36	int32_t code = (*pWrapper->func.requiredFp)(&input, &required);	3,776,920✔
37	if (!required) {	3,776,920✔
38	dDebug("node:%s, does not require startup", pWrapper->name);	1,795,860✔
39	} else {
40	dDebug("node:%s, required to startup", pWrapper->name);	1,981,060✔
41	}
42
43	return required;	3,776,920✔
44	}
45
46	int32_t dmInitDnode(SDnode *pDnode) {	539,560✔
47	dDebug("start to create dnode");	539,560✔
48	int32_t code = -1;	539,560✔
49	char path[PATH_MAX + 100] = {0};	539,560✔
50
51	if ((code = dmInitVarsWrapper(pDnode)) != 0) {	539,560✔
52	goto _OVER;	×
53	}
54
55	// compress module init
56	tsCompressInit(tsLossyColumns, tsFPrecision, tsDPrecision, tsMaxRange, tsCurRange, (int)tsIfAdtFse, tsCompressor);	539,560✔
57
58	pDnode->wrappers[DNODE].func = dmGetMgmtFunc();	539,560✔
59	pDnode->wrappers[MNODE].func = mmGetMgmtFunc();	539,560✔
60	pDnode->wrappers[VNODE].func = vmGetMgmtFunc();	539,560✔
61	pDnode->wrappers[QNODE].func = qmGetMgmtFunc();	539,560✔
62	pDnode->wrappers[SNODE].func = smGetMgmtFunc();	539,560✔
63	pDnode->wrappers[BNODE].func = bmGetMgmtFunc();	539,560✔
64	pDnode->wrappers[XNODE].func = xmGetMgmtFunc();	539,560✔
65
66	for (EDndNodeType ntype = DNODE; ntype < NODE_END; ++ntype) {	4,316,480✔
67	SMgmtWrapper *pWrapper = &pDnode->wrappers[ntype];	3,776,920✔
68	pWrapper->pDnode = pDnode;	3,776,920✔
69	pWrapper->name = dmNodeName(ntype);	3,776,920✔
70	pWrapper->ntype = ntype;	3,776,920✔
71	(void)taosThreadRwlockInit(&pWrapper->lock, NULL);	3,776,920✔
72
73	snprintf(path, sizeof(path), "%s%s%s", tsDataDir, TD_DIRSEP, pWrapper->name);	3,776,920✔
74	pWrapper->path = taosStrdup(path);	3,776,920✔
75	if (pWrapper->path == NULL) {	3,776,920✔
76	code = terrno;	×
77	goto _OVER;	×
78	}
79
80	pWrapper->required = dmRequireNode(pDnode, pWrapper);	3,776,920✔
81	}
82
83	code = dmCheckRunning(tsDataDir, &pDnode->lockfile);	539,560✔
84	if (code != 0) {	539,560✔
85	goto _OVER;	5,285✔
86	}
87
88	if ((code = dmInitModule(pDnode)) != 0) {	534,275✔
89	goto _OVER;	28✔
90	}
91
92	indexInit(tsNumOfCommitThreads);	534,247✔
93
94	if ((code = dmInitStatusClient(pDnode)) != 0) {	534,247✔
95	goto _OVER;	×
96	}
97	if ((code = dmInitSyncClient(pDnode)) != 0) {	534,247✔
98	goto _OVER;	×
99	}
100
101	dmReportStartup("dnode-transport", "initialized");	534,247✔
102	dDebug("dnode is created, ptr:%p", pDnode);	534,247✔
103	code = 0;	534,247✔
104
105	_OVER:	539,560✔
106	if (code != 0 && pDnode != NULL) {	539,560✔
107	dmClearVars(pDnode);	5,313✔
108	pDnode = NULL;	5,313✔
109	dError("failed to create dnode since %s", tstrerror(code));	5,313✔
110	}
111
112	return code;	539,560✔
113	}
114
115	void dmCleanupDnode(SDnode *pDnode) {	534,247✔
116	if (pDnode == NULL) {	534,247✔
117	return;	×
118	}
119
120	dmCleanupClient(pDnode);	534,247✔
121	dmCleanupStatusClient(pDnode);	534,247✔
122	dmCleanupSyncClient(pDnode);	534,247✔
123	dmCleanupServer(pDnode);	534,247✔
124
125	dmClearVars(pDnode);	534,247✔
126	rpcCleanup();	534,247✔
127	streamCleanup();	534,247✔
128	indexCleanup();	534,247✔
129	taosConvDestroy();	534,247✔
130
131	// compress destroy
132	tsCompressExit();	534,247✔
133
134	dDebug("dnode is closed, ptr:%p", pDnode);	534,247✔
135	}
136
137	int32_t dmInitVarsWrapper(SDnode *pDnode) {	539,560✔
138	int32_t code = dmInitVars(pDnode);	539,560✔
139	if (code == -1) {	539,560✔
140	return terrno;	×
141	}
142	return 0;	539,560✔
143	}
144	int32_t dmInitVars(SDnode *pDnode) {	539,560✔
145	int32_t code = 0;	539,560✔
146	SDnodeData *pData = &pDnode->data;	539,560✔
147	pData->dnodeId = 0;	539,560✔
148	pData->clusterId = 0;	539,560✔
149	pData->dnodeVer = 0;	539,560✔
150	pData->engineVer = 0;	539,560✔
151	pData->updateTime = 0;	539,560✔
152	pData->rebootTime = taosGetTimestampMs();	539,560✔
153	pData->dropped = 0;	539,560✔
154	pData->stopped = 0;	539,560✔
155	char *machineId = NULL;	539,560✔
156	code = tGetMachineId(&machineId);	539,560✔
157	if (machineId) {	539,560✔
158	tstrncpy(pData->machineId, machineId, TSDB_MACHINE_ID_LEN + 1);	539,560✔
159	taosMemoryFreeClear(machineId);	539,560✔
160	} else {
161	#if defined(TD_ENTERPRISE) && !defined(GRANTS_CFG)
162	code = TSDB_CODE_DNODE_NO_MACHINE_CODE;	×
163	return terrno = code;	×
164	#endif
165	}
166
167	pData->dnodeHash = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);	539,560✔
168	if (pData->dnodeHash == NULL) {	539,560✔
169	dError("failed to init dnode hash");	×
170	return terrno;	×
171	}
172
173	if ((code = dmReadEps(pData)) != 0) {	539,560✔
174	dError("failed to read file since %s", tstrerror(code));	×
175	return code;	×
176	}
177
178	#if defined(TD_ENTERPRISE) \|\| defined(TD_ASTRA_TODO)
179	tsiEncryptAlgorithm = pData->encryptAlgorigthm;	539,560✔
180	tsiEncryptScope = pData->encryptScope;	539,560✔
181
182	#if defined(TD_HAS_TAOSK) \|\| defined(TD_ASTRA_TODO)
183	// Load local encryption keys and initialize key version
184	{
185	char masterKeyFile[PATH_MAX] = {0};	539,560✔
186	char derivedKeyFile[PATH_MAX] = {0};	539,560✔
187	snprintf(masterKeyFile, sizeof(masterKeyFile), "%s%sdnode%sconfig%smaster.bin", tsDataDir, TD_DIRSEP, TD_DIRSEP,	539,560✔
188	TD_DIRSEP);
189	snprintf(derivedKeyFile, sizeof(derivedKeyFile), "%s%sdnode%sconfig%sderived.bin", tsDataDir, TD_DIRSEP, TD_DIRSEP,	539,560✔
190	TD_DIRSEP);
191
192	char svrKey[129] = {0};	539,560✔
193	char dbKey[129] = {0};	539,560✔
194	char cfgKey[129] = {0};	539,560✔
195	char metaKey[129] = {0};	539,560✔
196	char dataKey[129] = {0};	539,560✔
197	int32_t algorithm = 0;	539,560✔
198	int32_t cfgAlgorithm = 0;	539,560✔
199	int32_t metaAlgorithm = 0;	539,560✔
200	int32_t fileVersion = 0;	539,560✔
201	int32_t keyVersion = 0;	539,560✔
202	int64_t createTime = 0;	539,560✔
203	int64_t svrKeyUpdateTime = 0;	539,560✔
204	int64_t dbKeyUpdateTime = 0;	539,560✔
205
206	code = taoskLoadEncryptKeys(masterKeyFile, derivedKeyFile, svrKey, dbKey, cfgKey, metaKey, dataKey, &algorithm,	539,560✔
207	&cfgAlgorithm, &metaAlgorithm, &fileVersion, &keyVersion, &createTime,
208	&svrKeyUpdateTime, &dbKeyUpdateTime);
209	if (code == 0) {	539,560✔
210	tsLocalKeyVersion = keyVersion;	×
211	dInfo("loaded local encryption keys, version:%d", tsLocalKeyVersion);	×
212	} else {
213	tsLocalKeyVersion = 0;	539,560✔
214	dInfo("no local encryption keys found or failed to load, will sync from mnode");	539,560✔
215	}
216	}
217	#endif
218	#endif
219	/*
220	if(tsiEncryptAlgorithm != 0) {
221	if(pData->machineId != NULL && strlen(pData->machineId) > 0){
222	dInfo("get crypt key at startup, machineId:%s", pData->machineId);
223	int32_t code = 0;
224
225	//code = taosGetCryptKey(tsAuthCode, pData->machineId, tsCryptKey);
226	code = 0;
227	tstrncpy(tsEncryptKey, tsAuthCode, 16);
228
229	if (code != 0) {
230	if(code == -1){
231	terrno = TSDB_CODE_DNODE_NO_ENCRYPT_KEY;
232	dError("machine code changed, can't get crypt key");
233	}
234	if(code == -2){
235	terrno = TSDB_CODE_DNODE_NO_ENCRYPT_KEY;
236	dError("failed to get crypt key");
237	}
238	return -1;
239	}
240
241	if(strlen(tsEncryptKey) == 0){
242	terrno = TSDB_CODE_DNODE_NO_ENCRYPT_KEY;
243	dError("failed to get crypt key at startup since key is null, machineId:%s", pData->machineId);
244	return -1;
245	}
246	}
247	else{
248	terrno = TSDB_CODE_DNODE_NO_MACHINE_CODE;
249	dError("failed to get crypt key at startup, machineId:%s", pData->machineId);
250	return -1;
251	}
252	}
253	*/
254
255	if (pData->dropped) {	539,560✔
256	dError("dnode will not start since its already dropped");	×
257	return -1;	×
258	}
259
260	(void)taosThreadRwlockInit(&pData->lock, NULL);	539,560✔
261	(void)taosThreadMutexInit(&pData->statusInfolock, NULL);	539,560✔
262	(void)taosThreadMutexInit(&pDnode->mutex, NULL);	539,560✔
263	return 0;	539,560✔
264	}
265
266	extern SMonVloadInfo tsVinfo;
267
268	void dmClearVars(SDnode *pDnode) {	539,560✔
269	for (EDndNodeType ntype = DNODE; ntype < NODE_END; ++ntype) {	4,316,480✔
270	SMgmtWrapper *pWrapper = &pDnode->wrappers[ntype];	3,776,920✔
271	taosMemoryFreeClear(pWrapper->path);	3,776,920✔
272	(void)taosThreadRwlockDestroy(&pWrapper->lock);	3,776,920✔
273	}
274	if (pDnode->lockfile != NULL) {	539,560✔
275	if (taosUnLockFile(pDnode->lockfile) != 0) {	539,318✔
276	dError("failed to unlock file");	×
277	}
278
279	(void)taosCloseFile(&pDnode->lockfile);	539,318✔
280	pDnode->lockfile = NULL;	539,318✔
281	}
282
283	SDnodeData *pData = &pDnode->data;	539,560✔
284	(void)taosThreadRwlockWrlock(&pData->lock);	539,560✔
285	if (pData->oldDnodeEps != NULL) {	539,560✔
286	if (dmWriteEps(pData) == 0) {	×
287	dmRemoveDnodePairs(pData);	×
288	}
289	taosArrayDestroy(pData->oldDnodeEps);	×
290	pData->oldDnodeEps = NULL;	×
291	}
292	if (pData->dnodeEps != NULL) {	539,560✔
293	taosArrayDestroy(pData->dnodeEps);	539,560✔
294	pData->dnodeEps = NULL;	539,560✔
295	}
296	if (pData->dnodeHash != NULL) {	539,560✔
297	taosHashCleanup(pData->dnodeHash);	539,560✔
298	pData->dnodeHash = NULL;	539,560✔
299	}
300	(void)taosThreadRwlockUnlock(&pData->lock);	539,560✔
301
302	(void)taosThreadRwlockDestroy(&pData->lock);	539,560✔
303
304	dDebug("begin to lock status info when thread exit");	539,560✔
305	if (taosThreadMutexLock(&pData->statusInfolock) != 0) {	539,560✔
306	dError("failed to lock status info lock");	×
307	return;	×
308	}
309	if (tsVinfo.pVloads != NULL) {	539,560✔
310	taosArrayDestroy(tsVinfo.pVloads);	158,386✔
311	tsVinfo.pVloads = NULL;	158,386✔
312	}
313	if (taosThreadMutexUnlock(&pData->statusInfolock) != 0) {	539,560✔
314	dError("failed to unlock status info lock");	×
315	return;	×
316	}
317	if (taosThreadMutexDestroy(&pData->statusInfolock) != 0) {	539,560✔
318	dError("failed to destroy status info lock");	×
319	}
320	memset(&pData->statusInfolock, 0, sizeof(pData->statusInfolock));	539,560✔
321
322	(void)taosThreadMutexDestroy(&pDnode->mutex);	539,560✔
323	memset(&pDnode->mutex, 0, sizeof(pDnode->mutex));	539,560✔
324	}
325
326	void dmSetStatus(SDnode *pDnode, EDndRunStatus status) {	1,068,494✔
327	if (pDnode->status != status) {	1,068,494✔
328	dDebug("dnode status set from %s to %s", dmStatStr(pDnode->status), dmStatStr(status));	1,068,494✔
329	pDnode->status = status;	1,068,494✔
330	}
331	}	1,068,494✔
332
333	SMgmtWrapper dmAcquireWrapper(SDnode pDnode, EDndNodeType ntype) {	453,604✔
334	SMgmtWrapper *pWrapper = &pDnode->wrappers[ntype];	453,604✔
335	SMgmtWrapper *pRetWrapper = pWrapper;	453,604✔
336
337	(void)taosThreadRwlockRdlock(&pWrapper->lock);	453,604✔
338	if (pWrapper->deployed) {	453,604✔
339	int32_t refCount = atomic_add_fetch_32(&pWrapper->refCount, 1);	363,029✔
340	// dTrace("node:%s, is acquired, ref:%d", pWrapper->name, refCount);
341	} else {
342	pRetWrapper = NULL;	90,575✔
343	}
344	(void)taosThreadRwlockUnlock(&pWrapper->lock);	453,604✔
345
346	return pRetWrapper;	453,604✔
347	}
348
349	int32_t dmMarkWrapper(SMgmtWrapper *pWrapper) {	2,147,483,647✔
350	int32_t code = 0;	2,147,483,647✔
351
352	(void)taosThreadRwlockRdlock(&pWrapper->lock);	2,147,483,647✔
353	if (pWrapper->deployed) {	2,147,483,647✔
354	int32_t refCount = atomic_add_fetch_32(&pWrapper->refCount, 1);	2,147,483,647✔
355	// dTrace("node:%s, is marked, ref:%d", pWrapper->name, refCount);
356	} else {
357	switch (pWrapper->ntype) {	54,908,370✔
358	case MNODE:	15,962,925✔
359	code = TSDB_CODE_MNODE_NOT_FOUND;	15,962,925✔
360	break;	15,962,925✔
361	case QNODE:	38,830,247✔
362	code = TSDB_CODE_QNODE_NOT_FOUND;	38,830,247✔
363	break;	38,830,247✔
364	case SNODE:	492✔
365	code = TSDB_CODE_SNODE_NOT_FOUND;	492✔
366	break;	492✔
367	case BNODE:	×
368	code = TSDB_CODE_BNODE_NOT_FOUND;	×
369	break;	×
NEW 370	case XNODE:	×
NEW 371	code = TSDB_CODE_XNODE_NOT_FOUND;	×
NEW 372	break;	×
373	case VNODE:	114,378✔
374	code = TSDB_CODE_VND_STOPPED;	114,378✔
375	break;	114,378✔
376	default:	×
377	code = TSDB_CODE_APP_IS_STOPPING;	×
378	break;	×
379	}
380	}
381	(void)taosThreadRwlockUnlock(&pWrapper->lock);	2,147,483,647✔
382
383	return code;	2,147,483,647✔
384	}
385
386	void dmReleaseWrapper(SMgmtWrapper *pWrapper) {	2,147,483,647✔
387	if (pWrapper == NULL) return;	2,147,483,647✔
388
389	(void)taosThreadRwlockRdlock(&pWrapper->lock);	2,147,483,647✔
390	int32_t refCount = atomic_sub_fetch_32(&pWrapper->refCount, 1);	2,147,483,647✔
391	(void)taosThreadRwlockUnlock(&pWrapper->lock);	2,147,483,647✔
392	// dTrace("node:%s, is released, ref:%d", pWrapper->name, refCount);
393	}
394
395	static void dmGetServerStartupStatus(SDnode pDnode, SServerStatusRsp pStatus) {	×
396	SDnodeMgmt *pMgmt = pDnode->wrappers[DNODE].pMgmt;	×
397	pStatus->details[0] = 0;	×
398
399	if (pDnode->status == DND_STAT_INIT) {	×
400	pStatus->statusCode = TSDB_SRV_STATUS_NETWORK_OK;	×
401	snprintf(pStatus->details, sizeof(pStatus->details), "%s: %s", pDnode->startup.name, pDnode->startup.desc);	×
402	} else if (pDnode->status == DND_STAT_STOPPED) {	×
403	pStatus->statusCode = TSDB_SRV_STATUS_EXTING;	×
404	} else {
405	pStatus->statusCode = TSDB_SRV_STATUS_SERVICE_OK;	×
406	}
407	}	×
408
409	void dmProcessNetTestReq(SDnode pDnode, SRpcMsg pMsg) {	×
410	dDebug("msg:%p, net test req will be processed", pMsg);	×
411
412	SRpcMsg rsp = {.info = pMsg->info};	×
413	rsp.pCont = rpcMallocCont(pMsg->contLen);	×
414	if (rsp.pCont == NULL) {	×
415	rsp.code = TSDB_CODE_OUT_OF_MEMORY;	×
416	} else {
417	rsp.contLen = pMsg->contLen;	×
418	}
419
420	if (rpcSendResponse(&rsp) != 0) {	×
421	dError("failed to send response, msg:%p", &rsp);	×
422	}
423	rpcFreeCont(pMsg->pCont);	×
424	}	×
425
426	void dmProcessServerStartupStatus(SDnode pDnode, SRpcMsg pMsg) {	×
427	dDebug("msg:%p, server startup status req will be processed", pMsg);	×
428
429	SServerStatusRsp statusRsp = {0};	×
430	dmGetServerStartupStatus(pDnode, &statusRsp);	×
431
432	SRpcMsg rsp = {.info = pMsg->info};	×
433	int32_t contLen = tSerializeSServerStatusRsp(NULL, 0, &statusRsp);	×
434	if (contLen < 0) {	×
435	rsp.code = TSDB_CODE_OUT_OF_MEMORY;	×
436	} else {
437	rsp.pCont = rpcMallocCont(contLen);	×
438	if (rsp.pCont != NULL) {	×
439	if (tSerializeSServerStatusRsp(rsp.pCont, contLen, &statusRsp) < 0) {	×
440	rsp.code = TSDB_CODE_APP_ERROR;	×
441	} else {
442	rsp.contLen = contLen;	×
443	}
444	}
445	}
446
447	if (rpcSendResponse(&rsp) != 0) {	×
448	dError("failed to send response, msg:%p", &rsp);	×
449	}
450	rpcFreeCont(pMsg->pCont);	×
451	}	×

taosdata / TDengine / #4911

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