• Home
  • Features
  • Pricing
  • Docs
  • Announcements
  • Sign In

taosdata / TDengine / #4922

09 Jan 2026 08:13AM UTC coverage: 65.161% (-0.4%) from 65.541%
#4922

push

travis-ci

web-flow
merge: from main to 3.0 branch #34232

33 of 56 new or added lines in 8 files covered. (58.93%)

2171 existing lines in 120 files now uncovered.

197632 of 303297 relevant lines covered (65.16%)

117870313.81 hits per line

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

78.45
/source/dnode/mnode/impl/src/mndMnode.c
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 "audit.h"
18
#include "mndCluster.h"
19
#include "mndDnode.h"
20
#include "mndMnode.h"
21
#include "mndPrivilege.h"
22
#include "mndShow.h"
23
#include "mndSync.h"
24
#include "mndTrans.h"
25
#include "tmisce.h"
26

27
#define MNODE_VER_NUMBER   2
28
#define MNODE_RESERVE_SIZE 64
29

30
static int32_t  mndCreateDefaultMnode(SMnode *pMnode);
31
static SSdbRaw *mndMnodeActionEncode(SMnodeObj *pObj);
32
static SSdbRow *mndMnodeActionDecode(SSdbRaw *pRaw);
33
static int32_t  mndMnodeActionInsert(SSdb *pSdb, SMnodeObj *pObj);
34
static int32_t  mndMnodeActionDelete(SSdb *pSdb, SMnodeObj *pObj);
35
static int32_t  mndMnodeActionUpdate(SSdb *pSdb, SMnodeObj *pOld, SMnodeObj *pNew);
36
static int32_t  mndProcessCreateMnodeReq(SRpcMsg *pReq);
37
static int32_t  mndProcessAlterMnodeReq(SRpcMsg *pReq);
38
static int32_t  mndProcessDropMnodeReq(SRpcMsg *pReq);
39
static int32_t  mndRetrieveMnodes(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows);
40
static void     mndCancelGetNextMnode(SMnode *pMnode, void *pIter);
41
static void     mndReloadSyncConfig(SMnode *pMnode);
42

43
int32_t mndInitMnode(SMnode *pMnode) {
406,482✔
44
  SSdbTable table = {
406,482✔
45
      .sdbType = SDB_MNODE,
46
      .keyType = SDB_KEY_INT32,
47
      .deployFp = (SdbDeployFp)mndCreateDefaultMnode,
48
      .encodeFp = (SdbEncodeFp)mndMnodeActionEncode,
49
      .decodeFp = (SdbDecodeFp)mndMnodeActionDecode,
50
      .insertFp = (SdbInsertFp)mndMnodeActionInsert,
51
      .updateFp = (SdbUpdateFp)mndMnodeActionUpdate,
52
      .deleteFp = (SdbDeleteFp)mndMnodeActionDelete,
53
  };
54

55
  mndSetMsgHandle(pMnode, TDMT_MND_CREATE_MNODE, mndProcessCreateMnodeReq);
406,482✔
56
  mndSetMsgHandle(pMnode, TDMT_DND_CREATE_MNODE_RSP, mndTransProcessRsp);
406,482✔
57
  mndSetMsgHandle(pMnode, TDMT_DND_ALTER_MNODE_TYPE_RSP, mndTransProcessRsp);
406,482✔
58
  mndSetMsgHandle(pMnode, TDMT_MND_ALTER_MNODE, mndProcessAlterMnodeReq);
406,482✔
59
  mndSetMsgHandle(pMnode, TDMT_MND_ALTER_MNODE_RSP, mndTransProcessRsp);
406,482✔
60
  mndSetMsgHandle(pMnode, TDMT_MND_DROP_MNODE, mndProcessDropMnodeReq);
406,482✔
61
  mndSetMsgHandle(pMnode, TDMT_DND_DROP_MNODE_RSP, mndTransProcessRsp);
406,482✔
62

63
  mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_MNODE, mndRetrieveMnodes);
406,482✔
64
  mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_MNODE, mndCancelGetNextMnode);
406,482✔
65

66
  return sdbSetTable(pMnode->pSdb, table);
406,482✔
67
}
68

69
void mndCleanupMnode(SMnode *pMnode) {}
406,421✔
70

71
SMnodeObj *mndAcquireMnode(SMnode *pMnode, int32_t mnodeId) {
39,435,953✔
72
  terrno = 0;
39,435,953✔
73
  SMnodeObj *pObj = sdbAcquire(pMnode->pSdb, SDB_MNODE, &mnodeId);
39,435,953✔
74
  if (pObj == NULL && terrno == TSDB_CODE_SDB_OBJ_NOT_THERE) {
39,435,953✔
75
    terrno = TSDB_CODE_MND_MNODE_NOT_EXIST;
15,674,584✔
76
  }
77
  return pObj;
39,435,953✔
78
}
79

80
void mndReleaseMnode(SMnode *pMnode, SMnodeObj *pObj) {
23,720,091✔
81
  SSdb *pSdb = pMnode->pSdb;
23,720,091✔
82
  sdbRelease(pMnode->pSdb, pObj);
23,720,091✔
83
}
23,720,091✔
84

85
static int32_t mndCreateDefaultMnode(SMnode *pMnode) {
289,073✔
86
  int32_t   code = 0;
289,073✔
87
  SMnodeObj mnodeObj = {0};
289,073✔
88
  mnodeObj.id = 1;
289,073✔
89
  mnodeObj.createdTime = taosGetTimestampMs();
289,073✔
90
  mnodeObj.updateTime = mnodeObj.createdTime;
289,073✔
91

92
  SSdbRaw *pRaw = mndMnodeActionEncode(&mnodeObj);
289,073✔
93
  if (pRaw == NULL) {
289,073✔
94
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
95
    if (terrno != 0) code = terrno;
×
96
    return -1;
×
97
  }
98
  TAOS_CHECK_RETURN(sdbSetRawStatus(pRaw, SDB_STATUS_READY));
289,073✔
99

100
  mInfo("mnode:%d, will be created when deploying, raw:%p", mnodeObj.id, pRaw);
289,073✔
101

102
  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_GLOBAL, NULL, "create-mnode");
289,073✔
103
  if (pTrans == NULL) {
289,073✔
104
    sdbFreeRaw(pRaw);
×
105
    mError("mnode:%d, failed to create since %s", mnodeObj.id, terrstr());
×
106
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
107
    if (terrno != 0) code = terrno;
×
108
    return -1;
×
109
  }
110
  mInfo("trans:%d, used to create mnode:%d", pTrans->id, mnodeObj.id);
289,073✔
111

112
  if ((code = mndTransAppendCommitlog(pTrans, pRaw)) != 0) {
289,073✔
113
    mError("trans:%d, failed to append commit log since %s", pTrans->id, terrstr());
×
114
    mndTransDrop(pTrans);
×
115
    TAOS_RETURN(code);
×
116
  }
117
  TAOS_CHECK_RETURN(sdbSetRawStatus(pRaw, SDB_STATUS_READY));
289,073✔
118

119
  if ((code = mndTransPrepare(pMnode, pTrans)) != 0) {
289,073✔
120
    mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
×
121
    mndTransDrop(pTrans);
×
122
    return -1;
×
123
  }
124

125
  mndTransDrop(pTrans);
289,073✔
126
  TAOS_RETURN(code);
289,073✔
127
}
128

129
static SSdbRaw *mndMnodeActionEncode(SMnodeObj *pObj) {
1,186,667✔
130
  int32_t code = 0;
1,186,667✔
131
  int32_t lino = 0;
1,186,667✔
132
  terrno = TSDB_CODE_OUT_OF_MEMORY;
1,186,667✔
133

134
  SSdbRaw *pRaw = sdbAllocRaw(SDB_MNODE, MNODE_VER_NUMBER, sizeof(SMnodeObj) + MNODE_RESERVE_SIZE);
1,186,667✔
135
  if (pRaw == NULL) goto _OVER;
1,186,667✔
136

137
  int32_t dataPos = 0;
1,186,667✔
138
  SDB_SET_INT32(pRaw, dataPos, pObj->id, _OVER)
1,186,667✔
139
  SDB_SET_INT64(pRaw, dataPos, pObj->createdTime, _OVER)
1,186,667✔
140
  SDB_SET_INT64(pRaw, dataPos, pObj->updateTime, _OVER)
1,186,667✔
141
  SDB_SET_INT32(pRaw, dataPos, pObj->role, _OVER)
1,186,667✔
142
  SDB_SET_INT64(pRaw, dataPos, pObj->lastIndex, _OVER)
1,186,667✔
143
  SDB_SET_RESERVE(pRaw, dataPos, MNODE_RESERVE_SIZE, _OVER)
1,186,667✔
144

145
  terrno = 0;
1,186,667✔
146

147
_OVER:
1,186,667✔
148
  if (terrno != 0) {
1,186,667✔
149
    mError("mnode:%d, failed to encode to raw:%p since %s", pObj->id, pRaw, terrstr());
×
150
    sdbFreeRaw(pRaw);
×
151
    return NULL;
×
152
  }
153

154
  mTrace("mnode:%d, encode to raw:%p, row:%p", pObj->id, pRaw, pObj);
1,186,667✔
155
  return pRaw;
1,186,667✔
156
}
157

158
static SSdbRow *mndMnodeActionDecode(SSdbRaw *pRaw) {
534,754✔
159
  int32_t code = 0;
534,754✔
160
  int32_t lino = 0;
534,754✔
161
  terrno = TSDB_CODE_OUT_OF_MEMORY;
534,754✔
162
  SSdbRow   *pRow = NULL;
534,754✔
163
  SMnodeObj *pObj = NULL;
534,754✔
164

165
  int8_t sver = 0;
534,754✔
166
  if (sdbGetRawSoftVer(pRaw, &sver) != 0) return NULL;
534,754✔
167

168
  if (sver != 1 && sver != 2) {
534,754✔
169
    terrno = TSDB_CODE_SDB_INVALID_DATA_VER;
×
170
    goto _OVER;
×
171
  }
172

173
  pRow = sdbAllocRow(sizeof(SMnodeObj));
534,754✔
174
  if (pRow == NULL) goto _OVER;
534,754✔
175

176
  pObj = sdbGetRowObj(pRow);
534,754✔
177
  if (pObj == NULL) goto _OVER;
534,754✔
178

179
  int32_t dataPos = 0;
534,754✔
180
  SDB_GET_INT32(pRaw, dataPos, &pObj->id, _OVER)
534,754✔
181
  SDB_GET_INT64(pRaw, dataPos, &pObj->createdTime, _OVER)
534,754✔
182
  SDB_GET_INT64(pRaw, dataPos, &pObj->updateTime, _OVER)
534,754✔
183
  if (sver >= 2) {
534,754✔
184
    SDB_GET_INT32(pRaw, dataPos, &pObj->role, _OVER)
534,754✔
185
    SDB_GET_INT64(pRaw, dataPos, &pObj->lastIndex, _OVER)
534,754✔
186
  }
187
  SDB_GET_RESERVE(pRaw, dataPos, MNODE_RESERVE_SIZE, _OVER)
534,754✔
188

189
  terrno = 0;
534,754✔
190

191
_OVER:
534,754✔
192
  if (terrno != 0) {
534,754✔
193
    mError("mnode:%d, failed to decode from raw:%p since %s", pObj == NULL ? 0 : pObj->id, pRaw, terrstr());
×
194
    taosMemoryFreeClear(pRow);
×
195
    return NULL;
×
196
  }
197

198
  mTrace("mnode:%d, decode from raw:%p, row:%p", pObj->id, pRaw, pObj);
534,754✔
199
  return pRow;
534,754✔
200
}
201

202
static int32_t mndMnodeActionInsert(SSdb *pSdb, SMnodeObj *pObj) {
498,337✔
203
  int32_t code = 0;
498,337✔
204
  mTrace("mnode:%d, perform insert action, row:%p", pObj->id, pObj);
498,337✔
205
  pObj->pDnode = sdbAcquireNotReadyObj(pSdb, SDB_DNODE, &pObj->id);
498,337✔
206
  if (pObj->pDnode == NULL) {
498,337✔
207
    mError("mnode:%d, failed to perform insert action since %s", pObj->id, terrstr());
×
208
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
209
    if (terrno != 0) code = terrno;
×
210
    int32_t code = 0;
×
211
  }
212

213
  pObj->syncState = TAOS_SYNC_STATE_OFFLINE;
498,337✔
214
  mndReloadSyncConfig(pSdb->pMnode);
498,337✔
215
  TAOS_RETURN(code);
498,337✔
216
}
217

218
static int32_t mndMnodeActionDelete(SSdb *pSdb, SMnodeObj *pObj) {
534,708✔
219
  mTrace("mnode:%d, perform delete action, row:%p", pObj->id, pObj);
534,708✔
220
  if (pObj->pDnode != NULL) {
534,708✔
221
    sdbRelease(pSdb, pObj->pDnode);
498,291✔
222
    pObj->pDnode = NULL;
498,291✔
223
  }
224

225
  return 0;
534,708✔
226
}
227

228
static int32_t mndMnodeActionUpdate(SSdb *pSdb, SMnodeObj *pOld, SMnodeObj *pNew) {
34,945✔
229
  mTrace("mnode:%d, perform update action, old row:%p new row:%p", pOld->id, pOld, pNew);
34,945✔
230
  pOld->role = pNew->role;
34,945✔
231
  pOld->updateTime = pNew->updateTime;
34,945✔
232
  pOld->lastIndex = pNew->lastIndex;
34,945✔
233
  mndReloadSyncConfig(pSdb->pMnode);
34,945✔
234

235
  return 0;
34,945✔
236
}
237

238
bool mndIsMnode(SMnode *pMnode, int32_t dnodeId) {
108,654,040✔
239
  SSdb *pSdb = pMnode->pSdb;
108,654,040✔
240

241
  SMnodeObj *pObj = sdbAcquire(pSdb, SDB_MNODE, &dnodeId);
108,654,040✔
242
  if (pObj == NULL) {
108,654,040✔
243
    return false;
76,757,733✔
244
  }
245

246
  sdbRelease(pSdb, pObj);
31,896,307✔
247
  return true;
31,896,307✔
248
}
249

250
void mndGetMnodeEpSet(SMnode *pMnode, SEpSet *pEpSet) {
26,125,214✔
251
  if (pMnode == NULL || pEpSet == NULL) {
26,125,214✔
UNCOV
252
    return;
×
253
  }
254

255
  syncGetRetryEpSet(pMnode->syncMgmt.sync, pEpSet);
26,127,890✔
256

257
  /*
258
  SSdb   *pSdb = pMnode->pSdb;
259
  int32_t totalMnodes = sdbGetSize(pSdb, SDB_MNODE);
260
  if (totalMnodes == 0) {
261
    syncGetRetryEpSet(pMnode->syncMgmt.sync, pEpSet);
262
    return;
263
  }
264

265
  void *pIter = NULL;
266
  while (1) {
267
    SMnodeObj *pObj = NULL;
268
    pIter = sdbFetch(pSdb, SDB_MNODE, pIter, (void **)&pObj);
269
    if (pIter == NULL) break;
270

271
    if (pObj->id == pMnode->selfDnodeId) {
272
      if (mndIsLeader(pMnode)) {
273
        pEpSet->inUse = pEpSet->numOfEps;
274
      } else {
275
        pEpSet->inUse = (pEpSet->numOfEps + 1) % totalMnodes;
276
        // pEpSet->inUse = 0;
277
      }
278
    }
279
    if (pObj->pDnode != NULL) {
280
      if (addEpIntoEpSet(pEpSet, pObj->pDnode->fqdn, pObj->pDnode->port) != 0) {
281
        mError("mnode:%d, failed to add ep:%s:%d into epset", pObj->id, pObj->pDnode->fqdn, pObj->pDnode->port);
282
      }
283
      sdbRelease(pSdb, pObj);
284
    }
285

286
    if (pEpSet->numOfEps == 0) {
287
      syncGetRetryEpSet(pMnode->syncMgmt.sync, pEpSet);
288
    }
289

290
    if (pEpSet->inUse >= pEpSet->numOfEps) {
291
      pEpSet->inUse = 0;
292
    }
293
    epsetSort(pEpSet);
294
  }
295
    */
296
}
297

298
static int32_t mndSetCreateMnodeRedoLogs(SMnode *pMnode, STrans *pTrans, SMnodeObj *pObj) {
14,940✔
299
  int32_t  code = 0;
14,940✔
300
  SSdbRaw *pRedoRaw = mndMnodeActionEncode(pObj);
14,940✔
301
  if (pRedoRaw == NULL) {
14,940✔
302
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
303
    if (terrno != 0) code = terrno;
×
304
    TAOS_RETURN(code);
×
305
  }
306
  TAOS_CHECK_RETURN(mndTransAppendRedolog(pTrans, pRedoRaw));
14,940✔
307
  TAOS_CHECK_RETURN(sdbSetRawStatus(pRedoRaw, SDB_STATUS_CREATING));
14,940✔
308
  TAOS_RETURN(code);
14,940✔
309
}
310

311
int32_t mndSetRestoreCreateMnodeRedoLogs(SMnode *pMnode, STrans *pTrans, SMnodeObj *pObj) {
×
312
  int32_t  code = 0;
×
313
  SSdbRaw *pRedoRaw = mndMnodeActionEncode(pObj);
×
314
  if (pRedoRaw == NULL) {
×
315
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
316
    if (terrno != 0) code = terrno;
×
317
    TAOS_RETURN(code);
×
318
  }
319
  TAOS_CHECK_RETURN(mndTransAppendRedolog(pTrans, pRedoRaw));
×
320
  TAOS_CHECK_RETURN(sdbSetRawStatus(pRedoRaw, SDB_STATUS_READY));
×
321
  TAOS_RETURN(code);
×
322
}
323

324
static int32_t mndSetCreateMnodeUndoLogs(SMnode *pMnode, STrans *pTrans, SMnodeObj *pObj) {
×
325
  int32_t  code = 0;
×
326
  SSdbRaw *pUndoRaw = mndMnodeActionEncode(pObj);
×
327
  if (pUndoRaw == NULL) {
×
328
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
329
    if (terrno != 0) code = terrno;
×
330
    TAOS_RETURN(code);
×
331
  }
332
  TAOS_CHECK_RETURN(mndTransAppendUndolog(pTrans, pUndoRaw));
×
333
  TAOS_CHECK_RETURN(sdbSetRawStatus(pUndoRaw, SDB_STATUS_DROPPED));
×
334
  TAOS_RETURN(code);
×
335
}
336

337
int32_t mndSetCreateMnodeCommitLogs(SMnode *pMnode, STrans *pTrans, SMnodeObj *pObj) {
15,174✔
338
  int32_t  code = 0;
15,174✔
339
  SSdbRaw *pCommitRaw = mndMnodeActionEncode(pObj);
15,174✔
340
  if (pCommitRaw == NULL) {
15,174✔
341
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
342
    if (terrno != 0) code = terrno;
×
343
    TAOS_RETURN(code);
×
344
  }
345
  TAOS_CHECK_RETURN(mndTransAppendCommitlog(pTrans, pCommitRaw));
15,174✔
346
  TAOS_CHECK_RETURN(sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY));
15,174✔
347
  TAOS_RETURN(code);
15,174✔
348
}
349

350
static int32_t mndBuildCreateMnodeRedoAction(STrans *pTrans, SDCreateMnodeReq *pCreateReq, SEpSet *pCreateEpSet) {
15,174✔
351
  int32_t code = 0;
15,174✔
352
  int32_t contLen = tSerializeSDCreateMnodeReq(NULL, 0, pCreateReq);
15,174✔
353
  void   *pReq = taosMemoryMalloc(contLen);
15,174✔
354
  if (pReq == NULL) {
15,174✔
355
    code = terrno;
×
356
    return code;
×
357
  }
358
  code = tSerializeSDCreateMnodeReq(pReq, contLen, pCreateReq);
15,174✔
359
  if (code < 0) {
15,174✔
360
    taosMemoryFree(pReq);
×
361
    TAOS_RETURN(code);
×
362
  }
363

364
  STransAction action = {
15,174✔
365
      .epSet = *pCreateEpSet,
366
      .pCont = pReq,
367
      .contLen = contLen,
368
      .msgType = TDMT_DND_CREATE_MNODE,
369
      .acceptableCode = TSDB_CODE_MNODE_ALREADY_DEPLOYED,
370
      .groupId = -1,
371
  };
372

373
  if ((code = mndTransAppendRedoAction(pTrans, &action)) != 0) {
15,174✔
374
    taosMemoryFree(pReq);
×
375
    TAOS_RETURN(code);
×
376
  }
377
  TAOS_RETURN(code);
15,174✔
378
}
379

380
static int32_t mndBuildAlterMnodeTypeRedoAction(STrans *pTrans, SDAlterMnodeTypeReq *pAlterMnodeTypeReq,
15,174✔
381
                                                SEpSet *pAlterMnodeTypeEpSet) {
382
  int32_t code = 0;
15,174✔
383
  int32_t contLen = tSerializeSDCreateMnodeReq(NULL, 0, pAlterMnodeTypeReq);
15,174✔
384
  void   *pReq = taosMemoryMalloc(contLen);
15,174✔
385
  if (pReq == NULL) {
15,174✔
386
    code = terrno;
×
387
    return code;
×
388
  }
389
  code = tSerializeSDCreateMnodeReq(pReq, contLen, pAlterMnodeTypeReq);
15,174✔
390
  if (code < 0) {
15,174✔
391
    taosMemoryFree(pReq);
×
392
    TAOS_RETURN(code);
×
393
  }
394

395
  STransAction action = {
15,174✔
396
      .epSet = *pAlterMnodeTypeEpSet,
397
      .pCont = pReq,
398
      .contLen = contLen,
399
      .msgType = TDMT_DND_ALTER_MNODE_TYPE,
400
      .retryCode = TSDB_CODE_MNODE_NOT_CATCH_UP,
401
      .acceptableCode = TSDB_CODE_MNODE_ALREADY_IS_VOTER,
402
      .groupId = -1,
403
  };
404

405
  if ((code = mndTransAppendRedoAction(pTrans, &action)) != 0) {
15,174✔
406
    taosMemoryFree(pReq);
×
407
    TAOS_RETURN(code);
×
408
  }
409
  TAOS_RETURN(code);
15,174✔
410
}
411

412
static int32_t mndBuildAlterMnodeRedoAction(STrans *pTrans, SDCreateMnodeReq *pAlterReq, SEpSet *pAlterEpSet) {
×
413
  int32_t code = 0;
×
414
  int32_t contLen = tSerializeSDCreateMnodeReq(NULL, 0, pAlterReq);
×
415
  void   *pReq = taosMemoryMalloc(contLen);
×
416
  if (pReq == NULL) {
×
417
    code = terrno;
×
418
    return code;
×
419
  }
420
  code = tSerializeSDCreateMnodeReq(pReq, contLen, pAlterReq);
×
421
  if (code < 0) {
×
422
    taosMemoryFree(pReq);
×
423
    TAOS_RETURN(code);
×
424
  }
425
  STransAction action = {
×
426
      .epSet = *pAlterEpSet,
427
      .pCont = pReq,
428
      .contLen = contLen,
429
      .msgType = TDMT_MND_ALTER_MNODE,
430
      .acceptableCode = 0,
431
  };
432

433
  if ((code = mndTransAppendRedoAction(pTrans, &action)) != 0) {
×
434
    taosMemoryFree(pReq);
×
435
    TAOS_RETURN(code);
×
436
  }
437

438
  TAOS_RETURN(code);
×
439
}
440

441
static int32_t mndBuildDropMnodeRedoAction(STrans *pTrans, SDDropMnodeReq *pDropReq, SEpSet *pDroprEpSet) {
645✔
442
  int32_t code = 0;
645✔
443
  int32_t contLen = tSerializeSCreateDropMQSNodeReq(NULL, 0, pDropReq);
645✔
444
  void   *pReq = taosMemoryMalloc(contLen);
645✔
445
  if (pReq == NULL) {
645✔
446
    code = terrno;
×
447
    return code;
×
448
  }
449
  code = tSerializeSCreateDropMQSNodeReq(pReq, contLen, pDropReq);
645✔
450
  if (code < 0) {
645✔
451
    taosMemoryFree(pReq);
×
452
    TAOS_RETURN(code);
×
453
  }
454

455
  STransAction action = {
645✔
456
      .epSet = *pDroprEpSet,
457
      .pCont = pReq,
458
      .contLen = contLen,
459
      .msgType = TDMT_DND_DROP_MNODE,
460
      .acceptableCode = TSDB_CODE_MNODE_NOT_DEPLOYED,
461
      .groupId = -1,
462
  };
463

464
  if ((code = mndTransAppendRedoAction(pTrans, &action)) != 0) {
645✔
465
    taosMemoryFree(pReq);
×
466
    TAOS_RETURN(code);
×
467
  }
468
  TAOS_RETURN(code);
645✔
469
}
470

471
static int32_t mndSetCreateMnodeRedoActions(SMnode *pMnode, STrans *pTrans, SDnodeObj *pDnode, SMnodeObj *pObj) {
14,940✔
472
  SSdb            *pSdb = pMnode->pSdb;
14,940✔
473
  void            *pIter = NULL;
14,940✔
474
  int32_t          numOfReplicas = 0;
14,940✔
475
  int32_t          numOfLearnerReplicas = 0;
14,940✔
476
  SDCreateMnodeReq createReq = {0};
14,940✔
477
  SEpSet           createEpset = {0};
14,940✔
478

479
  while (1) {
22,238✔
480
    SMnodeObj *pMObj = NULL;
37,178✔
481
    pIter = sdbFetch(pSdb, SDB_MNODE, pIter, (void **)&pMObj);
37,178✔
482
    if (pIter == NULL) break;
37,178✔
483

484
    if (pMObj->role == TAOS_SYNC_ROLE_VOTER) {
22,238✔
485
      createReq.replicas[numOfReplicas].id = pMObj->id;
22,238✔
486
      createReq.replicas[numOfReplicas].port = pMObj->pDnode->port;
22,238✔
487
      memcpy(createReq.replicas[numOfReplicas].fqdn, pMObj->pDnode->fqdn, TSDB_FQDN_LEN);
22,238✔
488
      numOfReplicas++;
22,238✔
489
    } else {
490
      createReq.learnerReplicas[numOfLearnerReplicas].id = pMObj->id;
×
491
      createReq.learnerReplicas[numOfLearnerReplicas].port = pMObj->pDnode->port;
×
492
      memcpy(createReq.learnerReplicas[numOfLearnerReplicas].fqdn, pMObj->pDnode->fqdn, TSDB_FQDN_LEN);
×
493
      numOfLearnerReplicas++;
×
494
    }
495

496
    sdbRelease(pSdb, pMObj);
22,238✔
497
  }
498

499
  createReq.replica = numOfReplicas;
14,940✔
500
  createReq.learnerReplica = numOfLearnerReplicas + 1;
14,940✔
501
  createReq.learnerReplicas[numOfLearnerReplicas].id = pDnode->id;
14,940✔
502
  createReq.learnerReplicas[numOfLearnerReplicas].port = pDnode->port;
14,940✔
503
  memcpy(createReq.learnerReplicas[numOfLearnerReplicas].fqdn, pDnode->fqdn, TSDB_FQDN_LEN);
14,940✔
504

505
  createReq.lastIndex = pObj->lastIndex;
14,940✔
506

507
  createEpset.inUse = 0;
14,940✔
508
  createEpset.numOfEps = 1;
14,940✔
509
  createEpset.eps[0].port = pDnode->port;
14,940✔
510
  memcpy(createEpset.eps[0].fqdn, pDnode->fqdn, TSDB_FQDN_LEN);
14,940✔
511

512
  TAOS_CHECK_RETURN(mndBuildCreateMnodeRedoAction(pTrans, &createReq, &createEpset));
14,940✔
513

514
  TAOS_RETURN(0);
14,940✔
515
}
516

517
int32_t mndSetRestoreCreateMnodeRedoActions(SMnode *pMnode, STrans *pTrans, SDnodeObj *pDnode, SMnodeObj *pObj) {
234✔
518
  SSdb            *pSdb = pMnode->pSdb;
234✔
519
  void            *pIter = NULL;
234✔
520
  SDCreateMnodeReq createReq = {0};
234✔
521
  SEpSet           createEpset = {0};
234✔
522

523
  while (1) {
702✔
524
    SMnodeObj *pMObj = NULL;
936✔
525
    pIter = sdbFetch(pSdb, SDB_MNODE, pIter, (void **)&pMObj);
936✔
526
    if (pIter == NULL) break;
936✔
527

528
    if (pMObj->id == pDnode->id) {
702✔
529
      sdbRelease(pSdb, pMObj);
234✔
530
      continue;
234✔
531
    }
532

533
    if (pMObj->role == TAOS_SYNC_ROLE_VOTER) {
468✔
534
      createReq.replicas[createReq.replica].id = pMObj->id;
468✔
535
      createReq.replicas[createReq.replica].port = pMObj->pDnode->port;
468✔
536
      memcpy(createReq.replicas[createReq.replica].fqdn, pMObj->pDnode->fqdn, TSDB_FQDN_LEN);
468✔
537
      createReq.replica++;
468✔
538
    } else {
539
      createReq.learnerReplicas[createReq.learnerReplica].id = pMObj->id;
×
540
      createReq.learnerReplicas[createReq.learnerReplica].port = pMObj->pDnode->port;
×
541
      memcpy(createReq.learnerReplicas[createReq.learnerReplica].fqdn, pMObj->pDnode->fqdn, TSDB_FQDN_LEN);
×
542
      createReq.learnerReplica++;
×
543
    }
544

545
    sdbRelease(pSdb, pMObj);
468✔
546
  }
547

548
  createReq.learnerReplicas[createReq.learnerReplica].id = pDnode->id;
234✔
549
  createReq.learnerReplicas[createReq.learnerReplica].port = pDnode->port;
234✔
550
  memcpy(createReq.learnerReplicas[createReq.learnerReplica].fqdn, pDnode->fqdn, TSDB_FQDN_LEN);
234✔
551
  createReq.learnerReplica++;
234✔
552

553
  createReq.lastIndex = pObj->lastIndex;
234✔
554

555
  createEpset.inUse = 0;
234✔
556
  createEpset.numOfEps = 1;
234✔
557
  createEpset.eps[0].port = pDnode->port;
234✔
558
  memcpy(createEpset.eps[0].fqdn, pDnode->fqdn, TSDB_FQDN_LEN);
234✔
559

560
  TAOS_CHECK_RETURN(mndBuildCreateMnodeRedoAction(pTrans, &createReq, &createEpset));
234✔
561

562
  TAOS_RETURN(0);
234✔
563
}
564

565
static int32_t mndSetAlterMnodeTypeRedoActions(SMnode *pMnode, STrans *pTrans, SDnodeObj *pDnode, SMnodeObj *pObj) {
14,940✔
566
  SSdb               *pSdb = pMnode->pSdb;
14,940✔
567
  void               *pIter = NULL;
14,940✔
568
  SDAlterMnodeTypeReq alterReq = {0};
14,940✔
569
  SEpSet              createEpset = {0};
14,940✔
570

571
  while (1) {
22,238✔
572
    SMnodeObj *pMObj = NULL;
37,178✔
573
    pIter = sdbFetch(pSdb, SDB_MNODE, pIter, (void **)&pMObj);
37,178✔
574
    if (pIter == NULL) break;
37,178✔
575

576
    if (pMObj->role == TAOS_SYNC_ROLE_VOTER) {
22,238✔
577
      alterReq.replicas[alterReq.replica].id = pMObj->id;
22,238✔
578
      alterReq.replicas[alterReq.replica].port = pMObj->pDnode->port;
22,238✔
579
      memcpy(alterReq.replicas[alterReq.replica].fqdn, pMObj->pDnode->fqdn, TSDB_FQDN_LEN);
22,238✔
580
      alterReq.replica++;
22,238✔
581
    } else {
582
      alterReq.learnerReplicas[alterReq.learnerReplica].id = pMObj->id;
×
583
      alterReq.learnerReplicas[alterReq.learnerReplica].port = pMObj->pDnode->port;
×
584
      memcpy(alterReq.learnerReplicas[alterReq.learnerReplica].fqdn, pMObj->pDnode->fqdn, TSDB_FQDN_LEN);
×
585
      alterReq.learnerReplica++;
×
586
    }
587

588
    sdbRelease(pSdb, pMObj);
22,238✔
589
  }
590

591
  alterReq.replicas[alterReq.replica].id = pDnode->id;
14,940✔
592
  alterReq.replicas[alterReq.replica].port = pDnode->port;
14,940✔
593
  memcpy(alterReq.replicas[alterReq.replica].fqdn, pDnode->fqdn, TSDB_FQDN_LEN);
14,940✔
594
  alterReq.replica++;
14,940✔
595

596
  alterReq.lastIndex = pObj->lastIndex;
14,940✔
597

598
  createEpset.inUse = 0;
14,940✔
599
  createEpset.numOfEps = 1;
14,940✔
600
  createEpset.eps[0].port = pDnode->port;
14,940✔
601
  memcpy(createEpset.eps[0].fqdn, pDnode->fqdn, TSDB_FQDN_LEN);
14,940✔
602

603
  TAOS_CHECK_RETURN(mndBuildAlterMnodeTypeRedoAction(pTrans, &alterReq, &createEpset));
14,940✔
604

605
  TAOS_RETURN(0);
14,940✔
606
}
607

608
int32_t mndSetRestoreAlterMnodeTypeRedoActions(SMnode *pMnode, STrans *pTrans, SDnodeObj *pDnode, SMnodeObj *pObj) {
234✔
609
  SSdb               *pSdb = pMnode->pSdb;
234✔
610
  void               *pIter = NULL;
234✔
611
  SDAlterMnodeTypeReq alterReq = {0};
234✔
612
  SEpSet              createEpset = {0};
234✔
613

614
  while (1) {
702✔
615
    SMnodeObj *pMObj = NULL;
936✔
616
    pIter = sdbFetch(pSdb, SDB_MNODE, pIter, (void **)&pMObj);
936✔
617
    if (pIter == NULL) break;
936✔
618

619
    if (pMObj->id == pDnode->id) {
702✔
620
      sdbRelease(pSdb, pMObj);
234✔
621
      continue;
234✔
622
    }
623

624
    if (pMObj->role == TAOS_SYNC_ROLE_VOTER) {
468✔
625
      alterReq.replicas[alterReq.replica].id = pMObj->id;
468✔
626
      alterReq.replicas[alterReq.replica].port = pMObj->pDnode->port;
468✔
627
      memcpy(alterReq.replicas[alterReq.replica].fqdn, pMObj->pDnode->fqdn, TSDB_FQDN_LEN);
468✔
628
      alterReq.replica++;
468✔
629
    } else {
630
      alterReq.learnerReplicas[alterReq.learnerReplica].id = pMObj->id;
×
631
      alterReq.learnerReplicas[alterReq.learnerReplica].port = pMObj->pDnode->port;
×
632
      memcpy(alterReq.learnerReplicas[alterReq.learnerReplica].fqdn, pMObj->pDnode->fqdn, TSDB_FQDN_LEN);
×
633
      alterReq.learnerReplica++;
×
634
    }
635

636
    sdbRelease(pSdb, pMObj);
468✔
637
  }
638

639
  alterReq.replicas[alterReq.replica].id = pDnode->id;
234✔
640
  alterReq.replicas[alterReq.replica].port = pDnode->port;
234✔
641
  memcpy(alterReq.replicas[alterReq.replica].fqdn, pDnode->fqdn, TSDB_FQDN_LEN);
234✔
642
  alterReq.replica++;
234✔
643

644
  alterReq.lastIndex = pObj->lastIndex;
234✔
645

646
  createEpset.inUse = 0;
234✔
647
  createEpset.numOfEps = 1;
234✔
648
  createEpset.eps[0].port = pDnode->port;
234✔
649
  memcpy(createEpset.eps[0].fqdn, pDnode->fqdn, TSDB_FQDN_LEN);
234✔
650

651
  TAOS_CHECK_RETURN(mndBuildAlterMnodeTypeRedoAction(pTrans, &alterReq, &createEpset));
234✔
652

653
  TAOS_RETURN(0);
234✔
654
}
655

656
static int32_t mndCreateMnode(SMnode *pMnode, SRpcMsg *pReq, SDnodeObj *pDnode, SMCreateMnodeReq *pCreate) {
14,940✔
657
  int32_t code = -1;
14,940✔
658

659
  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_GLOBAL, pReq, "create-mnode");
14,940✔
660
  if (pTrans == NULL) {
14,940✔
661
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
662
    if (terrno != 0) code = terrno;
×
663
    goto _OVER;
×
664
  }
665
  mndTransSetSerial(pTrans);
14,940✔
666
  mInfo("trans:%d, used to create mnode:%d", pTrans->id, pCreate->dnodeId);
14,940✔
667
  TAOS_CHECK_GOTO(mndTransCheckConflict(pMnode, pTrans), NULL, _OVER);
14,940✔
668

669
  SMnodeObj mnodeObj = {0};
14,940✔
670
  mnodeObj.id = pDnode->id;
14,940✔
671
  mnodeObj.createdTime = taosGetTimestampMs();
14,940✔
672
  mnodeObj.updateTime = mnodeObj.createdTime;
14,940✔
673
  mnodeObj.role = TAOS_SYNC_ROLE_LEARNER;
14,940✔
674
  mnodeObj.lastIndex = pMnode->applied;
14,940✔
675

676
  TAOS_CHECK_GOTO(mndSetCreateMnodeRedoActions(pMnode, pTrans, pDnode, &mnodeObj), NULL, _OVER);
14,940✔
677
  TAOS_CHECK_GOTO(mndSetCreateMnodeRedoLogs(pMnode, pTrans, &mnodeObj), NULL, _OVER);
14,940✔
678

679
  SMnodeObj mnodeLeaderObj = {0};
14,940✔
680
  mnodeLeaderObj.id = pDnode->id;
14,940✔
681
  mnodeLeaderObj.createdTime = taosGetTimestampMs();
14,940✔
682
  mnodeLeaderObj.updateTime = mnodeLeaderObj.createdTime;
14,940✔
683
  mnodeLeaderObj.role = TAOS_SYNC_ROLE_VOTER;
14,940✔
684
  mnodeLeaderObj.lastIndex = pMnode->applied + 1;
14,940✔
685

686
  TAOS_CHECK_GOTO(mndSetAlterMnodeTypeRedoActions(pMnode, pTrans, pDnode, &mnodeLeaderObj), NULL, _OVER);
14,940✔
687
  TAOS_CHECK_GOTO(mndSetCreateMnodeCommitLogs(pMnode, pTrans, &mnodeLeaderObj), NULL, _OVER);
14,940✔
688
  TAOS_CHECK_GOTO(mndTransPrepare(pMnode, pTrans), NULL, _OVER);
14,940✔
689

690
  code = 0;
14,940✔
691

692
_OVER:
14,940✔
693
  mndTransDrop(pTrans);
14,940✔
694
  TAOS_RETURN(code);
14,940✔
695
}
696

697
static int32_t mndProcessCreateMnodeReq(SRpcMsg *pReq) {
21,707✔
698
  SMnode          *pMnode = pReq->info.node;
21,707✔
699
  int32_t          code = -1;
21,707✔
700
  SMnodeObj       *pObj = NULL;
21,707✔
701
  SDnodeObj       *pDnode = NULL;
21,707✔
702
  SMCreateMnodeReq createReq = {0};
21,707✔
703
  int64_t          tss = taosGetTimestampMs();
21,707✔
704

705
  TAOS_CHECK_GOTO(tDeserializeSCreateDropMQSNodeReq(pReq->pCont, pReq->contLen, &createReq), NULL, _OVER);
21,707✔
706

707
  mInfo("mnode:%d, start to create", createReq.dnodeId);
21,707✔
708
  TAOS_CHECK_GOTO(mndCheckOperPrivilege(pMnode, RPC_MSG_USER(pReq), RPC_MSG_TOKEN(pReq), MND_OPER_CREATE_MNODE), NULL, _OVER);
21,707✔
709

710
  pObj = mndAcquireMnode(pMnode, createReq.dnodeId);
21,707✔
711
  if (pObj != NULL) {
21,707✔
712
    code = TSDB_CODE_MND_MNODE_ALREADY_EXIST;
4,521✔
713
    goto _OVER;
4,521✔
714
  } else if (terrno != TSDB_CODE_MND_MNODE_NOT_EXIST) {
17,186✔
715
    goto _OVER;
×
716
  }
717

718
  pDnode = mndAcquireDnode(pMnode, createReq.dnodeId);
17,186✔
719
  if (pDnode == NULL) {
17,186✔
720
    code = TSDB_CODE_MND_DNODE_NOT_EXIST;
1,056✔
721
    goto _OVER;
1,056✔
722
  }
723

724
  if (sdbGetSize(pMnode->pSdb, SDB_MNODE) >= 3) {
16,130✔
725
    code = TSDB_CODE_MND_TOO_MANY_MNODES;
264✔
726
    goto _OVER;
264✔
727
  }
728

729
  if (!mndIsDnodeOnline(pDnode, taosGetTimestampMs())) {
15,866✔
730
    code = TSDB_CODE_DNODE_OFFLINE;
926✔
731
    goto _OVER;
926✔
732
  }
733

734
  code = mndCreateMnode(pMnode, pReq, pDnode, &createReq);
14,940✔
735
  if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
14,940✔
736

737
  if (tsAuditLevel >= AUDIT_LEVEL_SYSTEM) {
14,940✔
738
    char    obj[40] = {0};
14,940✔
739
    int32_t bytes = snprintf(obj, sizeof(obj), "%d", createReq.dnodeId);
14,940✔
740
    if ((uint32_t)bytes < sizeof(obj)) {
14,940✔
741
      int64_t tse = taosGetTimestampMs();
14,940✔
742
      double  duration = (double)(tse - tss);
14,940✔
743
      duration = duration / 1000;
14,940✔
744
      auditRecord(pReq, pMnode->clusterId, "createMnode", "", obj, createReq.sql, createReq.sqlLen, duration, 0);
14,940✔
745
    } else {
746
      mError("mnode:%d, failed to audit create req since %s", createReq.dnodeId, tstrerror(TSDB_CODE_OUT_OF_RANGE));
×
747
    }
748
  }
749

750
_OVER:
21,707✔
751
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {
21,707✔
752
    mError("mnode:%d, failed to create since %s", createReq.dnodeId, terrstr());
6,767✔
753
  }
754

755
  mndReleaseMnode(pMnode, pObj);
21,707✔
756
  mndReleaseDnode(pMnode, pDnode);
21,707✔
757
  tFreeSMCreateQnodeReq(&createReq);
21,707✔
758

759
  TAOS_RETURN(code);
21,707✔
760
}
761

762
static int32_t mndSetDropMnodeRedoLogs(SMnode *pMnode, STrans *pTrans, SMnodeObj *pObj) {
645✔
763
  int32_t  code = 0;
645✔
764
  SSdbRaw *pRedoRaw = mndMnodeActionEncode(pObj);
645✔
765
  if (pRedoRaw == NULL) {
645✔
766
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
767
    if (terrno != 0) code = terrno;
×
768
    TAOS_RETURN(code);
×
769
  }
770
  TAOS_CHECK_RETURN(mndTransAppendGroupRedolog(pTrans, pRedoRaw, -1));
645✔
771
  TAOS_CHECK_RETURN(sdbSetRawStatus(pRedoRaw, SDB_STATUS_DROPPING));
645✔
772
  TAOS_RETURN(code);
645✔
773
}
774

775
static int32_t mndSetDropMnodeCommitLogs(SMnode *pMnode, STrans *pTrans, SMnodeObj *pObj) {
645✔
776
  int32_t  code = 0;
645✔
777
  SSdbRaw *pCommitRaw = mndMnodeActionEncode(pObj);
645✔
778
  if (pCommitRaw == NULL) {
645✔
779
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
780
    if (terrno != 0) code = terrno;
×
781
    TAOS_RETURN(code);
×
782
  }
783
  TAOS_CHECK_RETURN(mndTransAppendCommitlog(pTrans, pCommitRaw));
645✔
784
  TAOS_CHECK_RETURN(sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED));
645✔
785
  TAOS_RETURN(code);
645✔
786
}
787

788
static int32_t mndSetDropMnodeRedoActions(SMnode *pMnode, STrans *pTrans, SDnodeObj *pDnode, SMnodeObj *pObj,
645✔
789
                                          bool force) {
790
  int32_t        code = 0;
645✔
791
  SSdb          *pSdb = pMnode->pSdb;
645✔
792
  void          *pIter = NULL;
645✔
793
  SDDropMnodeReq dropReq = {0};
645✔
794
  SEpSet         dropEpSet = {0};
645✔
795

796
  dropReq.dnodeId = pDnode->id;
645✔
797
  dropEpSet.numOfEps = 1;
645✔
798
  dropEpSet.eps[0].port = pDnode->port;
645✔
799
  memcpy(dropEpSet.eps[0].fqdn, pDnode->fqdn, TSDB_FQDN_LEN);
645✔
800

801
  int32_t totalMnodes = sdbGetSize(pSdb, SDB_MNODE);
645✔
802
  if (totalMnodes == 2) {
645✔
803
    if (force) {
300✔
804
      mError("cant't force drop dnode, since a mnode on it and replica is 2");
×
805
      code = TSDB_CODE_MNODE_ONLY_TWO_MNODE;
×
806
      TAOS_RETURN(code);
×
807
    }
808
    mInfo("vgId:1, has %d mnodes, exec redo log first", totalMnodes);
300✔
809
    TAOS_CHECK_RETURN(mndSetDropMnodeRedoLogs(pMnode, pTrans, pObj));
300✔
810
    if (!force) {
300✔
811
      TAOS_CHECK_RETURN(mndBuildDropMnodeRedoAction(pTrans, &dropReq, &dropEpSet));
300✔
812
    }
813
  } else if (totalMnodes == 3) {
345✔
814
    mInfo("vgId:1, has %d mnodes, exec redo action first", totalMnodes);
345✔
815
    if (!force) {
345✔
816
      TAOS_CHECK_RETURN(mndBuildDropMnodeRedoAction(pTrans, &dropReq, &dropEpSet));
345✔
817
    }
818
    TAOS_CHECK_RETURN(mndSetDropMnodeRedoLogs(pMnode, pTrans, pObj));
345✔
819
  } else {
820
    TAOS_RETURN(-1);
×
821
  }
822

823
  TAOS_RETURN(code);
645✔
824
}
825

826
int32_t mndSetDropMnodeInfoToTrans(SMnode *pMnode, STrans *pTrans, SMnodeObj *pObj, bool force) {
645✔
827
  if (pObj == NULL) return 0;
645✔
828
  pObj->lastIndex = pMnode->applied;
645✔
829
  TAOS_CHECK_RETURN(mndSetDropMnodeRedoActions(pMnode, pTrans, pObj->pDnode, pObj, force));
645✔
830
  TAOS_CHECK_RETURN(mndSetDropMnodeCommitLogs(pMnode, pTrans, pObj));
645✔
831
  return 0;
645✔
832
}
833

834
static int32_t mndDropMnode(SMnode *pMnode, SRpcMsg *pReq, SMnodeObj *pObj) {
399✔
835
  int32_t code = -1;
399✔
836
  STrans *pTrans = NULL;
399✔
837

838
  pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_GLOBAL, pReq, "drop-mnode");
399✔
839
  if (pTrans == NULL) {
399✔
840
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
841
    if (terrno != 0) code = terrno;
×
842
    goto _OVER;
×
843
  }
844
  mndTransSetSerial(pTrans);
399✔
845
  mInfo("trans:%d, used to drop mnode:%d", pTrans->id, pObj->id);
399✔
846
  TAOS_CHECK_GOTO(mndTransCheckConflict(pMnode, pTrans), NULL, _OVER);
399✔
847

848
  TAOS_CHECK_GOTO(mndSetDropMnodeInfoToTrans(pMnode, pTrans, pObj, false), NULL, _OVER);
399✔
849
  TAOS_CHECK_GOTO(mndTransPrepare(pMnode, pTrans), NULL, _OVER);
399✔
850

851
  code = 0;
399✔
852

853
_OVER:
399✔
854
  mndTransDrop(pTrans);
399✔
855
  TAOS_RETURN(code);
399✔
856
}
857

858
static int32_t mndProcessDropMnodeReq(SRpcMsg *pReq) {
2,356✔
859
  SMnode        *pMnode = pReq->info.node;
2,356✔
860
  int32_t        code = -1;
2,356✔
861
  SMnodeObj     *pObj = NULL;
2,356✔
862
  SMDropMnodeReq dropReq = {0};
2,356✔
863
  int64_t        tss = taosGetTimestampMs();
2,356✔
864

865
  TAOS_CHECK_GOTO(tDeserializeSCreateDropMQSNodeReq(pReq->pCont, pReq->contLen, &dropReq), NULL, _OVER);
2,356✔
866

867
  mInfo("mnode:%d, start to drop", dropReq.dnodeId);
2,356✔
868
  TAOS_CHECK_GOTO(mndCheckOperPrivilege(pMnode, RPC_MSG_USER(pReq), RPC_MSG_TOKEN(pReq), MND_OPER_DROP_MNODE), NULL, _OVER);
2,356✔
869

870
  if (dropReq.dnodeId <= 0) {
2,356✔
871
    code = TSDB_CODE_INVALID_MSG;
×
872
    goto _OVER;
×
873
  }
874

875
  pObj = mndAcquireMnode(pMnode, dropReq.dnodeId);
2,356✔
876
  if (pObj == NULL) {
2,356✔
877
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
931✔
878
    if (terrno != 0) code = terrno;
931✔
879
    goto _OVER;
931✔
880
  }
881

882
  if (pMnode->selfDnodeId == dropReq.dnodeId) {
1,425✔
883
    code = TSDB_CODE_MND_CANT_DROP_LEADER;
760✔
884
    goto _OVER;
760✔
885
  }
886

887
  if (sdbGetSize(pMnode->pSdb, SDB_MNODE) <= 1) {
665✔
888
    code = TSDB_CODE_MND_TOO_FEW_MNODES;
×
889
    goto _OVER;
×
890
  }
891

892
  if (!mndIsDnodeOnline(pObj->pDnode, taosGetTimestampMs())) {
665✔
893
    code = TSDB_CODE_DNODE_OFFLINE;
266✔
894
    goto _OVER;
266✔
895
  }
896

897
  code = mndDropMnode(pMnode, pReq, pObj);
399✔
898
  if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
399✔
899

900
  if (tsAuditLevel >= AUDIT_LEVEL_SYSTEM) {
399✔
901
    char obj[40] = {0};
399✔
902
    (void)tsnprintf(obj, sizeof(obj), "%d", dropReq.dnodeId);
399✔
903

904
    int64_t tse = taosGetTimestampMs();
399✔
905
    double  duration = (double)(tse - tss);
399✔
906
    duration = duration / 1000;
399✔
907
    auditRecord(pReq, pMnode->clusterId, "dropMnode", "", obj, dropReq.sql, dropReq.sqlLen, duration, 0);
399✔
908
  }
909

910
_OVER:
2,356✔
911
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {
2,356✔
912
    mError("mnode:%d, failed to drop since %s", dropReq.dnodeId, terrstr());
1,957✔
913
  }
914

915
  mndReleaseMnode(pMnode, pObj);
2,356✔
916
  tFreeSMCreateQnodeReq(&dropReq);
2,356✔
917
  TAOS_RETURN(code);
2,356✔
918
}
919

920
static int32_t mndRetrieveMnodes(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {
302,796✔
921
  SMnode    *pMnode = pReq->info.node;
302,796✔
922
  SSdb      *pSdb = pMnode->pSdb;
302,796✔
923
  int32_t    numOfRows = 0;
302,796✔
924
  int32_t    cols = 0;
302,796✔
925
  SMnodeObj *pObj = NULL;
302,796✔
926
  SMnodeObj *pSelfObj = NULL;
302,796✔
927
  ESdbStatus objStatus = 0;
302,796✔
928
  char      *pWrite;
929
  int64_t    curMs = taosGetTimestampMs();
302,796✔
930
  int        code = 0;
302,796✔
931

932
  pSelfObj = sdbAcquire(pSdb, SDB_MNODE, &pMnode->selfDnodeId);
302,796✔
933
  if (pSelfObj == NULL) {
302,796✔
934
    mError("mnode:%d, failed to acquire self %s", pMnode->selfDnodeId, terrstr());
×
935
    goto _out;
×
936
  }
937

938
  while (numOfRows < rows) {
651,548✔
939
    pShow->pIter = sdbFetchAll(pSdb, SDB_MNODE, pShow->pIter, (void **)&pObj, &objStatus, true);
651,548✔
940
    if (pShow->pIter == NULL) break;
651,548✔
941

942
    cols = 0;
348,752✔
943
    SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
348,752✔
944
    code = colDataSetVal(pColInfo, numOfRows, (const char *)&pObj->id, false);
348,752✔
945
    if (code != 0) {
348,752✔
946
      mError("mnode:%d, failed to set col data val since %s", pObj->id, tstrerror(code));
×
947
      sdbRelease(pSdb, pObj);
×
948
      goto _out;
×
949
    }
950

951
    char b1[TSDB_EP_LEN + VARSTR_HEADER_SIZE] = {0};
348,752✔
952
    STR_WITH_MAXSIZE_TO_VARSTR(b1, pObj->pDnode->ep, TSDB_EP_LEN + VARSTR_HEADER_SIZE);
348,752✔
953

954
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
348,752✔
955
    code = colDataSetVal(pColInfo, numOfRows, b1, false);
348,752✔
956
    if (code != 0) {
348,752✔
957
      mError("mnode:%d, failed to set col data val since %s", pObj->id, tstrerror(code));
×
958
      sdbCancelFetch(pSdb, pShow->pIter);
×
959
      sdbRelease(pSdb, pObj);
×
960
      goto _out;
×
961
    }
962

963
    char role[20] = "offline";
348,752✔
964
    if (pObj->id == pMnode->selfDnodeId) {
348,752✔
965
      snprintf(role, sizeof(role), "%s%s", syncStr(TAOS_SYNC_STATE_LEADER), pMnode->restored ? "" : "*");
302,796✔
966
    }
967
    bool isDnodeOnline = mndIsDnodeOnline(pObj->pDnode, curMs);
348,752✔
968
    if (isDnodeOnline) {
348,752✔
969
      tstrncpy(role, syncStr(pObj->syncState), sizeof(role));
343,114✔
970
      if (pObj->syncState == TAOS_SYNC_STATE_LEADER && pObj->id != pMnode->selfDnodeId) {
343,114✔
971
        tstrncpy(role, syncStr(TAOS_SYNC_STATE_ERROR), sizeof(role));
×
972
        mError("mnode:%d, is leader too", pObj->id);
×
973
      }
974
    }
975
    char b2[12 + VARSTR_HEADER_SIZE] = {0};
348,752✔
976
    STR_WITH_MAXSIZE_TO_VARSTR(b2, role, pShow->pMeta->pSchemas[cols].bytes);
348,752✔
977
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
348,752✔
978
    code = colDataSetVal(pColInfo, numOfRows, (const char *)b2, false);
348,752✔
979
    if (code != 0) goto _err;
348,752✔
980

981
    const char *status = "ready";
348,752✔
982
    if (objStatus == SDB_STATUS_CREATING) status = "creating";
348,752✔
983
    if (objStatus == SDB_STATUS_DROPPING) status = "dropping";
348,752✔
984
    if (!isDnodeOnline) status = "offline";
348,752✔
985
    char b3[9 + VARSTR_HEADER_SIZE] = {0};
348,752✔
986
    STR_WITH_MAXSIZE_TO_VARSTR(b3, status, pShow->pMeta->pSchemas[cols].bytes);
348,752✔
987
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
348,752✔
988
    code = colDataSetVal(pColInfo, numOfRows, (const char *)b3, false);
348,752✔
989
    if (code != 0) goto _err;
348,752✔
990

991
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
348,752✔
992
    code = colDataSetVal(pColInfo, numOfRows, (const char *)&pObj->createdTime, false);
348,752✔
993
    if (code != 0) goto _err;
348,752✔
994

995
    int64_t roleTimeMs = (isDnodeOnline) ? pObj->roleTimeMs : 0;
348,752✔
996
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
348,752✔
997
    code = colDataSetVal(pColInfo, numOfRows, (const char *)&roleTimeMs, false);
348,752✔
998
    if (code != 0) goto _err;
348,752✔
999

1000
    numOfRows++;
348,752✔
1001
    sdbRelease(pSdb, pObj);
348,752✔
1002
  }
1003

1004
  pShow->numOfRows += numOfRows;
302,796✔
1005

1006
_out:
302,796✔
1007
  sdbRelease(pSdb, pSelfObj);
302,796✔
1008
  return numOfRows;
302,796✔
1009

1010
_err:
×
1011
  mError("mnode:%d, failed to set col data val since %s", pObj->id, tstrerror(code));
×
1012
  sdbCancelFetch(pSdb, pShow->pIter);
×
1013
  sdbRelease(pSdb, pObj);
×
1014
  sdbRelease(pSdb, pSelfObj);
×
1015
  return numOfRows;
×
1016
}
1017

1018
static void mndCancelGetNextMnode(SMnode *pMnode, void *pIter) {
×
1019
  SSdb *pSdb = pMnode->pSdb;
×
1020
  sdbCancelFetchByType(pSdb, pIter, SDB_MNODE);
×
1021
}
×
1022

1023
static int32_t mndProcessAlterMnodeReq(SRpcMsg *pReq) {
×
1024
#if 1
1025
  return 0;
×
1026
#else
1027
  int32_t         code = 0;
1028
  SMnode         *pMnode = pReq->info.node;
1029
  SDAlterMnodeReq alterReq = {0};
1030

1031
  TAOS_CHECK_RETURN(tDeserializeSDCreateMnodeReq(pReq->pCont, pReq->contLen, &alterReq));
1032

1033
  SMnodeOpt option = {.deploy = true, .numOfReplicas = alterReq.replica, .selfIndex = -1};
1034
  memcpy(option.replicas, alterReq.replicas, sizeof(alterReq.replicas));
1035
  for (int32_t i = 0; i < option.numOfReplicas; ++i) {
1036
    if (alterReq.replicas[i].id == pMnode->selfDnodeId) {
1037
      option.selfIndex = i;
1038
    }
1039
  }
1040

1041
  if (option.selfIndex == -1) {
1042
    mInfo("alter mnode not processed since selfIndex is -1", terrstr());
1043
    return 0;
1044
  }
1045

1046
  if ((code = mndWriteFile(pMnode->path, &option)) != 0) {
1047
    mError("failed to write mnode file since %s", terrstr());
1048
    TAOS_RETURN(code);
1049
  }
1050

1051
  SSyncCfg cfg = {.replicaNum = alterReq.replica, .myIndex = -1};
1052
  for (int32_t i = 0; i < alterReq.replica; ++i) {
1053
    SNodeInfo *pNode = &cfg.nodeInfo[i];
1054
    tstrncpy(pNode->nodeFqdn, alterReq.replicas[i].fqdn, sizeof(pNode->nodeFqdn));
1055
    pNode->nodePort = alterReq.replicas[i].port;
1056
    if (alterReq.replicas[i].id == pMnode->selfDnodeId) {
1057
      cfg.myIndex = i;
1058
    }
1059
  }
1060

1061
  if (cfg.myIndex == -1) {
1062
    mError("failed to alter mnode since myindex is -1");
1063
    return -1;
1064
  } else {
1065
    mInfo("start to alter mnode sync, replica:%d myIndex:%d", cfg.replicaNum, cfg.myIndex);
1066
    for (int32_t i = 0; i < alterReq.replica; ++i) {
1067
      SNodeInfo *pNode = &cfg.nodeInfo[i];
1068
      mInfo("index:%d, fqdn:%s port:%d", i, pNode->nodeFqdn, pNode->nodePort);
1069
    }
1070
  }
1071

1072
  code = syncReconfig(pMnode->syncMgmt.sync, &cfg);
1073
  if (code != 0) {
1074
    mError("failed to sync reconfig since %s", terrstr());
1075
  } else {
1076
    mInfo("alter mnode sync success");
1077
  }
1078

1079
  TAOS_RETURN(code);
1080
#endif
1081
}
1082

1083
static void mndReloadSyncConfig(SMnode *pMnode) {
533,282✔
1084
  SSdb      *pSdb = pMnode->pSdb;
533,282✔
1085
  SMnodeObj *pObj = NULL;
533,282✔
1086
  ESdbStatus objStatus = 0;
533,282✔
1087
  void      *pIter = NULL;
533,282✔
1088
  int32_t    updatingMnodes = 0;
533,282✔
1089
  int32_t    readyMnodes = 0;
533,282✔
1090
  int32_t    code = 0;
533,282✔
1091
  SSyncCfg   cfg = {
533,282✔
1092
        .myIndex = -1,
1093
        .lastIndex = 0,
1094
  };
1095
  SyncIndex maxIndex = 0;
533,282✔
1096

1097
  while (1) {
1098
    pIter = sdbFetchAll(pSdb, SDB_MNODE, pIter, (void **)&pObj, &objStatus, false);
1,252,916✔
1099
    if (pIter == NULL) break;
1,252,916✔
1100
    if (objStatus == SDB_STATUS_CREATING || objStatus == SDB_STATUS_DROPPING) {
719,634✔
1101
      mInfo("vgId:1, has updating mnode:%d, status:%s", pObj->id, sdbStatusName(objStatus));
48,077✔
1102
      updatingMnodes++;
48,077✔
1103
    }
1104
    if (objStatus == SDB_STATUS_READY) {
719,634✔
1105
      mInfo("vgId:1, has ready mnode:%d, status:%s", pObj->id, sdbStatusName(objStatus));
671,557✔
1106
      readyMnodes++;
671,557✔
1107
    }
1108

1109
    if (objStatus == SDB_STATUS_READY || objStatus == SDB_STATUS_CREATING) {
719,634✔
1110
      SNodeInfo *pNode = &cfg.nodeInfo[cfg.totalReplicaNum];
718,162✔
1111
      pNode->nodeId = pObj->pDnode->id;
718,162✔
1112
      pNode->clusterId = mndGetClusterId(pMnode);
718,162✔
1113
      pNode->nodePort = pObj->pDnode->port;
718,162✔
1114
      pNode->nodeRole = pObj->role;
718,162✔
1115
      tstrncpy(pNode->nodeFqdn, pObj->pDnode->fqdn, TSDB_FQDN_LEN);
718,162✔
1116
      code = tmsgUpdateDnodeInfo(&pNode->nodeId, &pNode->clusterId, pNode->nodeFqdn, &pNode->nodePort);
718,162✔
1117
      if (code != 0) {
718,162✔
1118
        mError("mnode:%d, failed to update dnode info since %s", pObj->id, terrstr());
×
1119
      }
1120
      mInfo("vgId:1, ep:%s:%u dnode:%d", pNode->nodeFqdn, pNode->nodePort, pNode->nodeId);
718,162✔
1121
      if (pObj->pDnode->id == pMnode->selfDnodeId) {
718,162✔
1122
        cfg.myIndex = cfg.totalReplicaNum;
469,296✔
1123
      }
1124
      if (pNode->nodeRole == TAOS_SYNC_ROLE_VOTER) {
718,162✔
1125
        cfg.replicaNum++;
671,557✔
1126
      }
1127
      cfg.totalReplicaNum++;
718,162✔
1128
      if (pObj->lastIndex > cfg.lastIndex) {
718,162✔
1129
        cfg.lastIndex = pObj->lastIndex;
182,587✔
1130
      }
1131
    }
1132

1133
    if (objStatus == SDB_STATUS_DROPPING) {
719,634✔
1134
      if (pObj->lastIndex > cfg.lastIndex) {
1,472✔
1135
        cfg.lastIndex = pObj->lastIndex;
1,472✔
1136
      }
1137
    }
1138

1139
    mInfo("vgId:1, mnode:%d, role:%d, lastIndex:%" PRId64, pObj->id, pObj->role, pObj->lastIndex);
719,634✔
1140

1141
    sdbReleaseLock(pSdb, pObj, false);
719,634✔
1142
  }
1143

1144
  // if (readyMnodes <= 0 || updatingMnodes <= 0) {
1145
  //   mInfo("vgId:1, mnode sync not reconfig since readyMnodes:%d updatingMnodes:%d", readyMnodes, updatingMnodes);
1146
  //   return;
1147
  // }
1148

1149
  if (cfg.myIndex == -1) {
533,282✔
1150
#if 1
1151
    mInfo("vgId:1, mnode sync not reconfig since selfIndex is -1");
63,986✔
1152
#else
1153
    // cannot reconfig because the leader may fail to elect after reboot
1154
    mInfo("vgId:1, mnode sync not reconfig since selfIndex is -1, do sync stop oper");
1155
    syncStop(pMnode->syncMgmt.sync);
1156
#endif
1157
    return;
63,986✔
1158
  }
1159

1160
  if (pMnode->syncMgmt.sync > 0) {
469,296✔
1161
    mInfo("vgId:1, mnode sync reconfig, totalReplica:%d replica:%d myIndex:%d", cfg.totalReplicaNum, cfg.replicaNum,
369,512✔
1162
          cfg.myIndex);
1163

1164
    for (int32_t i = 0; i < cfg.totalReplicaNum; ++i) {
865,354✔
1165
      SNodeInfo *pNode = &cfg.nodeInfo[i];
495,842✔
1166
      mInfo("vgId:1, index:%d, ep:%s:%u dnode:%d cluster:%" PRId64 " role:%d", i, pNode->nodeFqdn, pNode->nodePort,
495,842✔
1167
            pNode->nodeId, pNode->clusterId, pNode->nodeRole);
1168
    }
1169

1170
    int32_t code = syncReconfig(pMnode->syncMgmt.sync, &cfg);
369,512✔
1171
    if (code != 0) {
369,512✔
1172
      mError("vgId:1, mnode sync reconfig failed since %s", terrstr());
×
1173
    } else {
1174
      mInfo("vgId:1, mnode sync reconfig success");
369,512✔
1175
    }
1176
  }
1177
}
STATUS · Troubleshooting · Open an Issue · Sales · Support · CAREERS · ENTERPRISE · START FREE · SCHEDULE DEMO
ANNOUNCEMENTS · TWITTER · TOS & SLA · Supported CI Services · What's a CI service? · Automated Testing

© 2026 Coveralls, Inc