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

taosdata / TDengine / #4720

08 Sep 2025 08:43AM UTC coverage: 58.139% (-0.6%) from 58.762%
#4720

push

travis-ci

web-flow
Merge pull request #32881 from taosdata/enh/add-new-windows-ci

fix(ci): update workflow reference to use new Windows CI YAML

133181 of 292179 branches covered (45.58%)

Branch coverage included in aggregate %.

201691 of 283811 relevant lines covered (71.07%)

5442780.71 hits per line

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

65.54
/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) {
1,929✔
44
  SSdbTable table = {
1,929✔
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);
1,929✔
56
  mndSetMsgHandle(pMnode, TDMT_DND_CREATE_MNODE_RSP, mndTransProcessRsp);
1,929✔
57
  mndSetMsgHandle(pMnode, TDMT_DND_ALTER_MNODE_TYPE_RSP, mndTransProcessRsp);
1,929✔
58
  mndSetMsgHandle(pMnode, TDMT_MND_ALTER_MNODE, mndProcessAlterMnodeReq);
1,929✔
59
  mndSetMsgHandle(pMnode, TDMT_MND_ALTER_MNODE_RSP, mndTransProcessRsp);
1,929✔
60
  mndSetMsgHandle(pMnode, TDMT_MND_DROP_MNODE, mndProcessDropMnodeReq);
1,929✔
61
  mndSetMsgHandle(pMnode, TDMT_DND_DROP_MNODE_RSP, mndTransProcessRsp);
1,929✔
62

63
  mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_MNODE, mndRetrieveMnodes);
1,929✔
64
  mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_MNODE, mndCancelGetNextMnode);
1,929✔
65

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

69
void mndCleanupMnode(SMnode *pMnode) {}
1,929✔
70

71
SMnodeObj *mndAcquireMnode(SMnode *pMnode, int32_t mnodeId) {
95,596✔
72
  terrno = 0;
95,596✔
73
  SMnodeObj *pObj = sdbAcquire(pMnode->pSdb, SDB_MNODE, &mnodeId);
95,596✔
74
  if (pObj == NULL && terrno == TSDB_CODE_SDB_OBJ_NOT_THERE) {
95,596✔
75
    terrno = TSDB_CODE_MND_MNODE_NOT_EXIST;
32,341✔
76
  }
77
  return pObj;
95,596✔
78
}
79

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

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

92
  SSdbRaw *pRaw = mndMnodeActionEncode(&mnodeObj);
1,426✔
93
  if (pRaw == NULL) {
1,426!
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));
1,426!
99

100
  mInfo("mnode:%d, will be created when deploying, raw:%p", mnodeObj.id, pRaw);
1,426!
101

102
  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_GLOBAL, NULL, "create-mnode");
1,426✔
103
  if (pTrans == NULL) {
1,426!
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);
1,426!
111

112
  if ((code = mndTransAppendCommitlog(pTrans, pRaw)) != 0) {
1,426!
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));
1,426!
118

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

125
  mndTransDrop(pTrans);
1,426✔
126
  TAOS_RETURN(code);
1,426✔
127
}
128

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

134
  SSdbRaw *pRaw = sdbAllocRaw(SDB_MNODE, MNODE_VER_NUMBER, sizeof(SMnodeObj) + MNODE_RESERVE_SIZE);
5,805✔
135
  if (pRaw == NULL) goto _OVER;
5,805!
136

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

145
  terrno = 0;
5,805✔
146

147
_OVER:
5,805✔
148
  if (terrno != 0) {
5,805!
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);
5,805✔
155
  return pRaw;
5,805✔
156
}
157

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

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

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

173
  pRow = sdbAllocRow(sizeof(SMnodeObj));
2,762✔
174
  if (pRow == NULL) goto _OVER;
2,762!
175

176
  pObj = sdbGetRowObj(pRow);
2,762✔
177
  if (pObj == NULL) goto _OVER;
2,762!
178

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

189
  terrno = 0;
2,762✔
190

191
_OVER:
2,762✔
192
  if (terrno != 0) {
2,762!
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);
2,762✔
199
  return pRow;
2,762✔
200
}
201

202
static int32_t mndMnodeActionInsert(SSdb *pSdb, SMnodeObj *pObj) {
2,528✔
203
  int32_t code = 0;
2,528✔
204
  mTrace("mnode:%d, perform insert action, row:%p", pObj->id, pObj);
2,528✔
205
  pObj->pDnode = sdbAcquireNotReadyObj(pSdb, SDB_DNODE, &pObj->id);
2,528✔
206
  if (pObj->pDnode == NULL) {
2,528!
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;
2,528✔
214
  mndReloadSyncConfig(pSdb->pMnode);
2,528✔
215
  TAOS_RETURN(code);
2,528✔
216
}
217

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

225
  return 0;
2,762✔
226
}
227

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

235
  return 0;
222✔
236
}
237

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

241
  SMnodeObj *pObj = sdbAcquire(pSdb, SDB_MNODE, &dnodeId);
287,654✔
242
  if (pObj == NULL) {
287,654✔
243
    return false;
181,237✔
244
  }
245

246
  sdbRelease(pSdb, pObj);
106,417✔
247
  return true;
106,417✔
248
}
249

250
void mndGetMnodeEpSet(SMnode *pMnode, SEpSet *pEpSet) {
71,962✔
251
  if (pMnode == NULL || pEpSet == NULL) {
71,962!
252
    return;
×
253
  }
254

255
  syncGetRetryEpSet(pMnode->syncMgmt.sync, pEpSet);
71,967✔
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) {
90✔
299
  int32_t  code = 0;
90✔
300
  SSdbRaw *pRedoRaw = mndMnodeActionEncode(pObj);
90✔
301
  if (pRedoRaw == NULL) {
90!
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));
90!
307
  TAOS_CHECK_RETURN(sdbSetRawStatus(pRedoRaw, SDB_STATUS_CREATING));
90!
308
  TAOS_RETURN(code);
90✔
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) {
92✔
338
  int32_t  code = 0;
92✔
339
  SSdbRaw *pCommitRaw = mndMnodeActionEncode(pObj);
92✔
340
  if (pCommitRaw == NULL) {
92!
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));
92!
346
  TAOS_CHECK_RETURN(sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY));
92!
347
  TAOS_RETURN(code);
92✔
348
}
349

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

364
  STransAction action = {
92✔
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) {
92!
374
    taosMemoryFree(pReq);
×
375
    TAOS_RETURN(code);
×
376
  }
377
  TAOS_RETURN(code);
92✔
378
}
379

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

395
  STransAction action = {
92✔
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) {
92!
406
    taosMemoryFree(pReq);
×
407
    TAOS_RETURN(code);
×
408
  }
409
  TAOS_RETURN(code);
92✔
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) {
5✔
442
  int32_t code = 0;
5✔
443
  int32_t contLen = tSerializeSCreateDropMQSNodeReq(NULL, 0, pDropReq);
5✔
444
  void   *pReq = taosMemoryMalloc(contLen);
5!
445
  if (pReq == NULL) {
5!
446
    code = terrno;
×
447
    return code;
×
448
  }
449
  code = tSerializeSCreateDropMQSNodeReq(pReq, contLen, pDropReq);
5✔
450
  if (code < 0) {
5!
451
    taosMemoryFree(pReq);
×
452
    TAOS_RETURN(code);
×
453
  }
454

455
  STransAction action = {
5✔
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) {
5!
465
    taosMemoryFree(pReq);
×
466
    TAOS_RETURN(code);
×
467
  }
468
  TAOS_RETURN(code);
5✔
469
}
470

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

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

484
    if (pMObj->role == TAOS_SYNC_ROLE_VOTER) {
134!
485
      createReq.replicas[numOfReplicas].id = pMObj->id;
134✔
486
      createReq.replicas[numOfReplicas].port = pMObj->pDnode->port;
134✔
487
      memcpy(createReq.replicas[numOfReplicas].fqdn, pMObj->pDnode->fqdn, TSDB_FQDN_LEN);
134✔
488
      numOfReplicas++;
134✔
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);
134✔
497
  }
498

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

505
  createReq.lastIndex = pObj->lastIndex;
90✔
506

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

512
  TAOS_CHECK_RETURN(mndBuildCreateMnodeRedoAction(pTrans, &createReq, &createEpset));
90!
513

514
  TAOS_RETURN(0);
90✔
515
}
516

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

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

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

533
    if (pMObj->role == TAOS_SYNC_ROLE_VOTER) {
4!
534
      createReq.replicas[createReq.replica].id = pMObj->id;
4✔
535
      createReq.replicas[createReq.replica].port = pMObj->pDnode->port;
4✔
536
      memcpy(createReq.replicas[createReq.replica].fqdn, pMObj->pDnode->fqdn, TSDB_FQDN_LEN);
4✔
537
      createReq.replica++;
4✔
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);
4✔
546
  }
547

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

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

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

560
  TAOS_CHECK_RETURN(mndBuildCreateMnodeRedoAction(pTrans, &createReq, &createEpset));
2!
561

562
  TAOS_RETURN(0);
2✔
563
}
564

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

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

576
    if (pMObj->role == TAOS_SYNC_ROLE_VOTER) {
134!
577
      alterReq.replicas[alterReq.replica].id = pMObj->id;
134✔
578
      alterReq.replicas[alterReq.replica].port = pMObj->pDnode->port;
134✔
579
      memcpy(alterReq.replicas[alterReq.replica].fqdn, pMObj->pDnode->fqdn, TSDB_FQDN_LEN);
134✔
580
      alterReq.replica++;
134✔
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);
134✔
589
  }
590

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

596
  alterReq.lastIndex = pObj->lastIndex;
90✔
597

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

603
  TAOS_CHECK_RETURN(mndBuildAlterMnodeTypeRedoAction(pTrans, &alterReq, &createEpset));
90!
604

605
  TAOS_RETURN(0);
90✔
606
}
607

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

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

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

624
    if (pMObj->role == TAOS_SYNC_ROLE_VOTER) {
4!
625
      alterReq.replicas[alterReq.replica].id = pMObj->id;
4✔
626
      alterReq.replicas[alterReq.replica].port = pMObj->pDnode->port;
4✔
627
      memcpy(alterReq.replicas[alterReq.replica].fqdn, pMObj->pDnode->fqdn, TSDB_FQDN_LEN);
4✔
628
      alterReq.replica++;
4✔
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);
4✔
637
  }
638

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

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

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

651
  TAOS_CHECK_RETURN(mndBuildAlterMnodeTypeRedoAction(pTrans, &alterReq, &createEpset));
2!
652

653
  TAOS_RETURN(0);
2✔
654
}
655

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

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

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

676
  TAOS_CHECK_GOTO(mndSetCreateMnodeRedoActions(pMnode, pTrans, pDnode, &mnodeObj), NULL, _OVER);
90!
677
  TAOS_CHECK_GOTO(mndSetCreateMnodeRedoLogs(pMnode, pTrans, &mnodeObj), NULL, _OVER);
90!
678

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

686
  TAOS_CHECK_GOTO(mndSetAlterMnodeTypeRedoActions(pMnode, pTrans, pDnode, &mnodeLeaderObj), NULL, _OVER);
90!
687
  TAOS_CHECK_GOTO(mndSetCreateMnodeCommitLogs(pMnode, pTrans, &mnodeLeaderObj), NULL, _OVER);
90!
688
  TAOS_CHECK_GOTO(mndTransPrepare(pMnode, pTrans), NULL, _OVER);
90!
689

690
  code = 0;
90✔
691

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

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

704
  TAOS_CHECK_GOTO(tDeserializeSCreateDropMQSNodeReq(pReq->pCont, pReq->contLen, &createReq), NULL, _OVER);
146!
705

706
  mInfo("mnode:%d, start to create", createReq.dnodeId);
146!
707
  TAOS_CHECK_GOTO(mndCheckOperPrivilege(pMnode, pReq->info.conn.user, MND_OPER_CREATE_MNODE), NULL, _OVER);
146✔
708

709
  pObj = mndAcquireMnode(pMnode, createReq.dnodeId);
145✔
710
  if (pObj != NULL) {
145✔
711
    code = TSDB_CODE_MND_MNODE_ALREADY_EXIST;
38✔
712
    goto _OVER;
38✔
713
  } else if (terrno != TSDB_CODE_MND_MNODE_NOT_EXIST) {
107!
714
    goto _OVER;
×
715
  }
716

717
  pDnode = mndAcquireDnode(pMnode, createReq.dnodeId);
107✔
718
  if (pDnode == NULL) {
107✔
719
    code = TSDB_CODE_MND_DNODE_NOT_EXIST;
8✔
720
    goto _OVER;
8✔
721
  }
722

723
  if (sdbGetSize(pMnode->pSdb, SDB_MNODE) >= 3) {
99✔
724
    code = TSDB_CODE_MND_TOO_MANY_MNODES;
2✔
725
    goto _OVER;
2✔
726
  }
727

728
  if (!mndIsDnodeOnline(pDnode, taosGetTimestampMs())) {
97✔
729
    code = TSDB_CODE_DNODE_OFFLINE;
7✔
730
    goto _OVER;
7✔
731
  }
732

733
  code = mndCreateMnode(pMnode, pReq, pDnode, &createReq);
90✔
734
  if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
90!
735

736
  char    obj[40] = {0};
90✔
737
  int32_t bytes = snprintf(obj, sizeof(obj), "%d", createReq.dnodeId);
90✔
738
  if ((uint32_t)bytes < sizeof(obj)) {
90!
739
    auditRecord(pReq, pMnode->clusterId, "createMnode", "", obj, createReq.sql, createReq.sqlLen);
90✔
740
  } else {
741
    mError("mnode:%d, failed to audit create req since %s", createReq.dnodeId, tstrerror(TSDB_CODE_OUT_OF_RANGE));
×
742
  }
743

744
_OVER:
×
745
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {
146!
746
    mError("mnode:%d, failed to create since %s", createReq.dnodeId, terrstr());
56!
747
  }
748

749
  mndReleaseMnode(pMnode, pObj);
146✔
750
  mndReleaseDnode(pMnode, pDnode);
146✔
751
  tFreeSMCreateQnodeReq(&createReq);
146✔
752

753
  TAOS_RETURN(code);
146✔
754
}
755

756
static int32_t mndSetDropMnodeRedoLogs(SMnode *pMnode, STrans *pTrans, SMnodeObj *pObj) {
5✔
757
  int32_t  code = 0;
5✔
758
  SSdbRaw *pRedoRaw = mndMnodeActionEncode(pObj);
5✔
759
  if (pRedoRaw == NULL) {
5!
760
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
761
    if (terrno != 0) code = terrno;
×
762
    TAOS_RETURN(code);
×
763
  }
764
  TAOS_CHECK_RETURN(mndTransAppendGroupRedolog(pTrans, pRedoRaw, -1));
5!
765
  TAOS_CHECK_RETURN(sdbSetRawStatus(pRedoRaw, SDB_STATUS_DROPPING));
5!
766
  TAOS_RETURN(code);
5✔
767
}
768

769
static int32_t mndSetDropMnodeCommitLogs(SMnode *pMnode, STrans *pTrans, SMnodeObj *pObj) {
5✔
770
  int32_t  code = 0;
5✔
771
  SSdbRaw *pCommitRaw = mndMnodeActionEncode(pObj);
5✔
772
  if (pCommitRaw == NULL) {
5!
773
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
774
    if (terrno != 0) code = terrno;
×
775
    TAOS_RETURN(code);
×
776
  }
777
  TAOS_CHECK_RETURN(mndTransAppendCommitlog(pTrans, pCommitRaw));
5!
778
  TAOS_CHECK_RETURN(sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED));
5!
779
  TAOS_RETURN(code);
5✔
780
}
781

782
static int32_t mndSetDropMnodeRedoActions(SMnode *pMnode, STrans *pTrans, SDnodeObj *pDnode, SMnodeObj *pObj,
5✔
783
                                          bool force) {
784
  int32_t        code = 0;
5✔
785
  SSdb          *pSdb = pMnode->pSdb;
5✔
786
  void          *pIter = NULL;
5✔
787
  SDDropMnodeReq dropReq = {0};
5✔
788
  SEpSet         dropEpSet = {0};
5✔
789

790
  dropReq.dnodeId = pDnode->id;
5✔
791
  dropEpSet.numOfEps = 1;
5✔
792
  dropEpSet.eps[0].port = pDnode->port;
5✔
793
  memcpy(dropEpSet.eps[0].fqdn, pDnode->fqdn, TSDB_FQDN_LEN);
5✔
794

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

817
  TAOS_RETURN(code);
5✔
818
}
819

820
int32_t mndSetDropMnodeInfoToTrans(SMnode *pMnode, STrans *pTrans, SMnodeObj *pObj, bool force) {
5✔
821
  if (pObj == NULL) return 0;
5!
822
  pObj->lastIndex = pMnode->applied;
5✔
823
  TAOS_CHECK_RETURN(mndSetDropMnodeRedoActions(pMnode, pTrans, pObj->pDnode, pObj, force));
5!
824
  TAOS_CHECK_RETURN(mndSetDropMnodeCommitLogs(pMnode, pTrans, pObj));
5!
825
  return 0;
5✔
826
}
827

828
static int32_t mndDropMnode(SMnode *pMnode, SRpcMsg *pReq, SMnodeObj *pObj) {
3✔
829
  int32_t code = -1;
3✔
830
  STrans *pTrans = NULL;
3✔
831

832
  pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_GLOBAL, pReq, "drop-mnode");
3✔
833
  if (pTrans == NULL) {
3!
834
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
×
835
    if (terrno != 0) code = terrno;
×
836
    goto _OVER;
×
837
  }
838
  mndTransSetSerial(pTrans);
3✔
839
  mInfo("trans:%d, used to drop mnode:%d", pTrans->id, pObj->id);
3!
840
  TAOS_CHECK_GOTO(mndTransCheckConflict(pMnode, pTrans), NULL, _OVER);
3!
841

842
  TAOS_CHECK_GOTO(mndSetDropMnodeInfoToTrans(pMnode, pTrans, pObj, false), NULL, _OVER);
3!
843
  TAOS_CHECK_GOTO(mndTransPrepare(pMnode, pTrans), NULL, _OVER);
3!
844

845
  code = 0;
3✔
846

847
_OVER:
3✔
848
  mndTransDrop(pTrans);
3✔
849
  TAOS_RETURN(code);
3✔
850
}
851

852
static int32_t mndProcessDropMnodeReq(SRpcMsg *pReq) {
19✔
853
  SMnode        *pMnode = pReq->info.node;
19✔
854
  int32_t        code = -1;
19✔
855
  SMnodeObj     *pObj = NULL;
19✔
856
  SMDropMnodeReq dropReq = {0};
19✔
857

858
  TAOS_CHECK_GOTO(tDeserializeSCreateDropMQSNodeReq(pReq->pCont, pReq->contLen, &dropReq), NULL, _OVER);
19!
859

860
  mInfo("mnode:%d, start to drop", dropReq.dnodeId);
19!
861
  TAOS_CHECK_GOTO(mndCheckOperPrivilege(pMnode, pReq->info.conn.user, MND_OPER_DROP_MNODE), NULL, _OVER);
19✔
862

863
  if (dropReq.dnodeId <= 0) {
18!
864
    code = TSDB_CODE_INVALID_MSG;
×
865
    goto _OVER;
×
866
  }
867

868
  pObj = mndAcquireMnode(pMnode, dropReq.dnodeId);
18✔
869
  if (pObj == NULL) {
18✔
870
    code = TSDB_CODE_MND_RETURN_VALUE_NULL;
7✔
871
    if (terrno != 0) code = terrno;
7!
872
    goto _OVER;
7✔
873
  }
874

875
  if (pMnode->selfDnodeId == dropReq.dnodeId) {
11✔
876
    code = TSDB_CODE_MND_CANT_DROP_LEADER;
6✔
877
    goto _OVER;
6✔
878
  }
879

880
  if (sdbGetSize(pMnode->pSdb, SDB_MNODE) <= 1) {
5!
881
    code = TSDB_CODE_MND_TOO_FEW_MNODES;
×
882
    goto _OVER;
×
883
  }
884

885
  if (!mndIsDnodeOnline(pObj->pDnode, taosGetTimestampMs())) {
5✔
886
    code = TSDB_CODE_DNODE_OFFLINE;
2✔
887
    goto _OVER;
2✔
888
  }
889

890
  code = mndDropMnode(pMnode, pReq, pObj);
3✔
891
  if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
3!
892

893
  char obj[40] = {0};
3✔
894
  (void)tsnprintf(obj, sizeof(obj), "%d", dropReq.dnodeId);
3✔
895

896
  auditRecord(pReq, pMnode->clusterId, "dropMnode", "", obj, dropReq.sql, dropReq.sqlLen);
3✔
897

898
_OVER:
19✔
899
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {
19!
900
    mError("mnode:%d, failed to drop since %s", dropReq.dnodeId, terrstr());
16!
901
  }
902

903
  mndReleaseMnode(pMnode, pObj);
19✔
904
  tFreeSMCreateQnodeReq(&dropReq);
19✔
905
  TAOS_RETURN(code);
19✔
906
}
907

908
static int32_t mndRetrieveMnodes(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {
1,550✔
909
  SMnode    *pMnode = pReq->info.node;
1,550✔
910
  SSdb      *pSdb = pMnode->pSdb;
1,550✔
911
  int32_t    numOfRows = 0;
1,550✔
912
  int32_t    cols = 0;
1,550✔
913
  SMnodeObj *pObj = NULL;
1,550✔
914
  SMnodeObj *pSelfObj = NULL;
1,550✔
915
  ESdbStatus objStatus = 0;
1,550✔
916
  char      *pWrite;
917
  int64_t    curMs = taosGetTimestampMs();
1,550✔
918
  int        code = 0;
1,550✔
919

920
  pSelfObj = sdbAcquire(pSdb, SDB_MNODE, &pMnode->selfDnodeId);
1,550✔
921
  if (pSelfObj == NULL) {
1,550!
922
    mError("mnode:%d, failed to acquire self %s", pMnode->selfDnodeId, terrstr());
×
923
    goto _out;
×
924
  }
925

926
  while (numOfRows < rows) {
3,420!
927
    pShow->pIter = sdbFetchAll(pSdb, SDB_MNODE, pShow->pIter, (void **)&pObj, &objStatus, true);
3,420✔
928
    if (pShow->pIter == NULL) break;
3,420✔
929

930
    cols = 0;
1,870✔
931
    SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
1,870✔
932
    code = colDataSetVal(pColInfo, numOfRows, (const char *)&pObj->id, false);
1,870✔
933
    if (code != 0) {
1,870!
934
      mError("mnode:%d, failed to set col data val since %s", pObj->id, tstrerror(code));
×
935
      sdbCancelFetch(pSdb, pShow->pIter);
×
936
      sdbRelease(pSdb, pObj);
×
937
      goto _out;
×
938
    }
939

940
    char b1[TSDB_EP_LEN + VARSTR_HEADER_SIZE] = {0};
1,870✔
941
    STR_WITH_MAXSIZE_TO_VARSTR(b1, pObj->pDnode->ep, TSDB_EP_LEN + VARSTR_HEADER_SIZE);
1,870✔
942

943
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
1,870✔
944
    code = colDataSetVal(pColInfo, numOfRows, b1, false);
1,870✔
945
    if (code != 0) {
1,870!
946
      mError("mnode:%d, failed to set col data val since %s", pObj->id, tstrerror(code));
×
947
      sdbCancelFetch(pSdb, pShow->pIter);
×
948
      sdbRelease(pSdb, pObj);
×
949
      goto _out;
×
950
    }
951

952
    char role[20] = "offline";
1,870✔
953
    if (pObj->id == pMnode->selfDnodeId) {
1,870✔
954
      snprintf(role, sizeof(role), "%s%s", syncStr(TAOS_SYNC_STATE_LEADER), pMnode->restored ? "" : "*");
1,550!
955
    }
956
    bool isDnodeOnline = mndIsDnodeOnline(pObj->pDnode, curMs);
1,870✔
957
    if (isDnodeOnline) {
1,870✔
958
      tstrncpy(role, syncStr(pObj->syncState), sizeof(role));
1,837✔
959
      if (pObj->syncState == TAOS_SYNC_STATE_LEADER && pObj->id != pMnode->selfDnodeId) {
1,837!
960
        tstrncpy(role, syncStr(TAOS_SYNC_STATE_ERROR), sizeof(role));
×
961
        mError("mnode:%d, is leader too", pObj->id);
×
962
      }
963
    }
964
    char b2[12 + VARSTR_HEADER_SIZE] = {0};
1,870✔
965
    STR_WITH_MAXSIZE_TO_VARSTR(b2, role, pShow->pMeta->pSchemas[cols].bytes);
1,870✔
966
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
1,870✔
967
    code = colDataSetVal(pColInfo, numOfRows, (const char *)b2, false);
1,870✔
968
    if (code != 0) goto _err;
1,870!
969

970
    const char *status = "ready";
1,870✔
971
    if (objStatus == SDB_STATUS_CREATING) status = "creating";
1,870!
972
    if (objStatus == SDB_STATUS_DROPPING) status = "dropping";
1,870!
973
    if (!isDnodeOnline) status = "offline";
1,870✔
974
    char b3[9 + VARSTR_HEADER_SIZE] = {0};
1,870✔
975
    STR_WITH_MAXSIZE_TO_VARSTR(b3, status, pShow->pMeta->pSchemas[cols].bytes);
1,870✔
976
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
1,870✔
977
    code = colDataSetVal(pColInfo, numOfRows, (const char *)b3, false);
1,870✔
978
    if (code != 0) goto _err;
1,870!
979

980
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
1,870✔
981
    code = colDataSetVal(pColInfo, numOfRows, (const char *)&pObj->createdTime, false);
1,870✔
982
    if (code != 0) goto _err;
1,870!
983

984
    int64_t roleTimeMs = (isDnodeOnline) ? pObj->roleTimeMs : 0;
1,870✔
985
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
1,870✔
986
    code = colDataSetVal(pColInfo, numOfRows, (const char *)&roleTimeMs, false);
1,870✔
987
    if (code != 0) goto _err;
1,870!
988

989
    numOfRows++;
1,870✔
990
    sdbRelease(pSdb, pObj);
1,870✔
991
  }
992

993
  pShow->numOfRows += numOfRows;
1,550✔
994

995
_out:
1,550✔
996
  sdbRelease(pSdb, pSelfObj);
1,550✔
997
  return numOfRows;
1,550✔
998

999
_err:
×
1000
  mError("mnode:%d, failed to set col data val since %s", pObj->id, tstrerror(code));
×
1001
  sdbCancelFetch(pSdb, pShow->pIter);
×
1002
  sdbRelease(pSdb, pObj);
×
1003
  sdbRelease(pSdb, pSelfObj);
×
1004
  return numOfRows;
×
1005
}
1006

1007
static void mndCancelGetNextMnode(SMnode *pMnode, void *pIter) {
×
1008
  SSdb *pSdb = pMnode->pSdb;
×
1009
  sdbCancelFetchByType(pSdb, pIter, SDB_MNODE);
×
1010
}
×
1011

1012
static int32_t mndProcessAlterMnodeReq(SRpcMsg *pReq) {
×
1013
#if 1
1014
  return 0;
×
1015
#else
1016
  int32_t         code = 0;
1017
  SMnode         *pMnode = pReq->info.node;
1018
  SDAlterMnodeReq alterReq = {0};
1019

1020
  TAOS_CHECK_RETURN(tDeserializeSDCreateMnodeReq(pReq->pCont, pReq->contLen, &alterReq));
1021

1022
  SMnodeOpt option = {.deploy = true, .numOfReplicas = alterReq.replica, .selfIndex = -1};
1023
  memcpy(option.replicas, alterReq.replicas, sizeof(alterReq.replicas));
1024
  for (int32_t i = 0; i < option.numOfReplicas; ++i) {
1025
    if (alterReq.replicas[i].id == pMnode->selfDnodeId) {
1026
      option.selfIndex = i;
1027
    }
1028
  }
1029

1030
  if (option.selfIndex == -1) {
1031
    mInfo("alter mnode not processed since selfIndex is -1", terrstr());
1032
    return 0;
1033
  }
1034

1035
  if ((code = mndWriteFile(pMnode->path, &option)) != 0) {
1036
    mError("failed to write mnode file since %s", terrstr());
1037
    TAOS_RETURN(code);
1038
  }
1039

1040
  SSyncCfg cfg = {.replicaNum = alterReq.replica, .myIndex = -1};
1041
  for (int32_t i = 0; i < alterReq.replica; ++i) {
1042
    SNodeInfo *pNode = &cfg.nodeInfo[i];
1043
    tstrncpy(pNode->nodeFqdn, alterReq.replicas[i].fqdn, sizeof(pNode->nodeFqdn));
1044
    pNode->nodePort = alterReq.replicas[i].port;
1045
    if (alterReq.replicas[i].id == pMnode->selfDnodeId) {
1046
      cfg.myIndex = i;
1047
    }
1048
  }
1049

1050
  if (cfg.myIndex == -1) {
1051
    mError("failed to alter mnode since myindex is -1");
1052
    return -1;
1053
  } else {
1054
    mInfo("start to alter mnode sync, replica:%d myIndex:%d", cfg.replicaNum, cfg.myIndex);
1055
    for (int32_t i = 0; i < alterReq.replica; ++i) {
1056
      SNodeInfo *pNode = &cfg.nodeInfo[i];
1057
      mInfo("index:%d, fqdn:%s port:%d", i, pNode->nodeFqdn, pNode->nodePort);
1058
    }
1059
  }
1060

1061
  code = syncReconfig(pMnode->syncMgmt.sync, &cfg);
1062
  if (code != 0) {
1063
    mError("failed to sync reconfig since %s", terrstr());
1064
  } else {
1065
    mInfo("alter mnode sync success");
1066
  }
1067

1068
  TAOS_RETURN(code);
1069
#endif
1070
}
1071

1072
static void mndReloadSyncConfig(SMnode *pMnode) {
2,750✔
1073
  SSdb      *pSdb = pMnode->pSdb;
2,750✔
1074
  SMnodeObj *pObj = NULL;
2,750✔
1075
  ESdbStatus objStatus = 0;
2,750✔
1076
  void      *pIter = NULL;
2,750✔
1077
  int32_t    updatingMnodes = 0;
2,750✔
1078
  int32_t    readyMnodes = 0;
2,750✔
1079
  int32_t    code = 0;
2,750✔
1080
  SSyncCfg   cfg = {
2,750✔
1081
        .myIndex = -1,
1082
        .lastIndex = 0,
1083
  };
1084
  SyncIndex maxIndex = 0;
2,750✔
1085

1086
  while (1) {
1087
    pIter = sdbFetchAll(pSdb, SDB_MNODE, pIter, (void **)&pObj, &objStatus, false);
6,708✔
1088
    if (pIter == NULL) break;
6,708✔
1089
    if (objStatus == SDB_STATUS_CREATING || objStatus == SDB_STATUS_DROPPING) {
3,958✔
1090
      mInfo("vgId:1, has updating mnode:%d, status:%s", pObj->id, sdbStatusName(objStatus));
305!
1091
      updatingMnodes++;
305✔
1092
    }
1093
    if (objStatus == SDB_STATUS_READY) {
3,958✔
1094
      mInfo("vgId:1, has ready mnode:%d, status:%s", pObj->id, sdbStatusName(objStatus));
3,653!
1095
      readyMnodes++;
3,653✔
1096
    }
1097

1098
    if (objStatus == SDB_STATUS_READY || objStatus == SDB_STATUS_CREATING) {
3,958✔
1099
      SNodeInfo *pNode = &cfg.nodeInfo[cfg.totalReplicaNum];
3,946✔
1100
      pNode->nodeId = pObj->pDnode->id;
3,946✔
1101
      pNode->clusterId = mndGetClusterId(pMnode);
3,946✔
1102
      pNode->nodePort = pObj->pDnode->port;
3,946✔
1103
      pNode->nodeRole = pObj->role;
3,946✔
1104
      tstrncpy(pNode->nodeFqdn, pObj->pDnode->fqdn, TSDB_FQDN_LEN);
3,946✔
1105
      code = tmsgUpdateDnodeInfo(&pNode->nodeId, &pNode->clusterId, pNode->nodeFqdn, &pNode->nodePort);
3,946✔
1106
      if (code != 0) {
3,946!
1107
        mError("mnode:%d, failed to update dnode info since %s", pObj->id, terrstr());
×
1108
      }
1109
      mInfo("vgId:1, ep:%s:%u dnode:%d", pNode->nodeFqdn, pNode->nodePort, pNode->nodeId);
3,946!
1110
      if (pObj->pDnode->id == pMnode->selfDnodeId) {
3,946✔
1111
        cfg.myIndex = cfg.totalReplicaNum;
2,340✔
1112
      }
1113
      if (pNode->nodeRole == TAOS_SYNC_ROLE_VOTER) {
3,946✔
1114
        cfg.replicaNum++;
3,653✔
1115
      }
1116
      cfg.totalReplicaNum++;
3,946✔
1117
      if (pObj->lastIndex > cfg.lastIndex) {
3,946✔
1118
        cfg.lastIndex = pObj->lastIndex;
1,180✔
1119
      }
1120
    }
1121

1122
    if (objStatus == SDB_STATUS_DROPPING) {
3,958✔
1123
      if (pObj->lastIndex > cfg.lastIndex) {
12!
1124
        cfg.lastIndex = pObj->lastIndex;
12✔
1125
      }
1126
    }
1127

1128
    mInfo("vgId:1, mnode:%d, role:%d, lastIndex:%" PRId64, pObj->id, pObj->role, pObj->lastIndex);
3,958!
1129

1130
    sdbReleaseLock(pSdb, pObj, false);
3,958✔
1131
  }
1132

1133
  // if (readyMnodes <= 0 || updatingMnodes <= 0) {
1134
  //   mInfo("vgId:1, mnode sync not reconfig since readyMnodes:%d updatingMnodes:%d", readyMnodes, updatingMnodes);
1135
  //   return;
1136
  // }
1137

1138
  if (cfg.myIndex == -1) {
2,750✔
1139
#if 1
1140
    mInfo("vgId:1, mnode sync not reconfig since selfIndex is -1");
410!
1141
#else
1142
    // cannot reconfig because the leader may fail to elect after reboot
1143
    mInfo("vgId:1, mnode sync not reconfig since selfIndex is -1, do sync stop oper");
1144
    syncStop(pMnode->syncMgmt.sync);
1145
#endif
1146
    return;
410✔
1147
  }
1148

1149
  if (pMnode->syncMgmt.sync > 0) {
2,340✔
1150
    mInfo("vgId:1, mnode sync reconfig, totalReplica:%d replica:%d myIndex:%d", cfg.totalReplicaNum, cfg.replicaNum,
1,927!
1151
          cfg.myIndex);
1152

1153
    for (int32_t i = 0; i < cfg.totalReplicaNum; ++i) {
4,637✔
1154
      SNodeInfo *pNode = &cfg.nodeInfo[i];
2,710✔
1155
      mInfo("vgId:1, index:%d, ep:%s:%u dnode:%d cluster:%" PRId64 " role:%d", i, pNode->nodeFqdn, pNode->nodePort,
2,710!
1156
            pNode->nodeId, pNode->clusterId, pNode->nodeRole);
1157
    }
1158

1159
    int32_t code = syncReconfig(pMnode->syncMgmt.sync, &cfg);
1,927✔
1160
    if (code != 0) {
1,927!
1161
      mError("vgId:1, mnode sync reconfig failed since %s", terrstr());
×
1162
    } else {
1163
      mInfo("vgId:1, mnode sync reconfig success");
1,927!
1164
    }
1165
  }
1166
}
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

© 2025 Coveralls, Inc