mndMnode.c 24.4 KB
Newer Older
H
refact  
Hongze Cheng 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15
/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

S
Shengliang Guan 已提交
16
#define _DEFAULT_SOURCE
S
Shengliang Guan 已提交
17 18
#include "mndMnode.h"
#include "mndDnode.h"
H
Hongze Cheng 已提交
19
#include "mndPrivilege.h"
S
Shengliang Guan 已提交
20
#include "mndShow.h"
21
#include "mndSync.h"
S
Shengliang Guan 已提交
22
#include "mndTrans.h"
S
Shengliang Guan 已提交
23
#include "mndUser.h"
S
Shengliang Guan 已提交
24

25 26
#define MNODE_VER_NUMBER   1
#define MNODE_RESERVE_SIZE 64
S
Shengliang Guan 已提交
27

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

int32_t mndInitMnode(SMnode *pMnode) {
S
Shengliang Guan 已提交
42 43 44 45 46 47 48 49 50 51
  SSdbTable table = {
      .sdbType = SDB_MNODE,
      .keyType = SDB_KEY_INT32,
      .deployFp = (SdbDeployFp)mndCreateDefaultMnode,
      .encodeFp = (SdbEncodeFp)mndMnodeActionEncode,
      .decodeFp = (SdbDecodeFp)mndMnodeActionDecode,
      .insertFp = (SdbInsertFp)mndMnodeActionInsert,
      .updateFp = (SdbUpdateFp)mndMnodeActionUpdate,
      .deleteFp = (SdbDeleteFp)mndMnodeActionDelete,
  };
S
Shengliang Guan 已提交
52

H
Hongze Cheng 已提交
53
  mndSetMsgHandle(pMnode, TDMT_MND_CREATE_MNODE, mndProcessCreateMnodeReq);
54
  mndSetMsgHandle(pMnode, TDMT_DND_CREATE_MNODE_RSP, mndTransProcessRsp);
S
Shengliang Guan 已提交
55
  mndSetMsgHandle(pMnode, TDMT_MND_ALTER_MNODE, mndProcessAlterMnodeReq);
56
  mndSetMsgHandle(pMnode, TDMT_MND_ALTER_MNODE_RSP, mndTransProcessRsp);
H
Hongze Cheng 已提交
57
  mndSetMsgHandle(pMnode, TDMT_MND_DROP_MNODE, mndProcessDropMnodeReq);
58
  mndSetMsgHandle(pMnode, TDMT_DND_DROP_MNODE_RSP, mndTransProcessRsp);
59 60
  mndSetMsgHandle(pMnode, TDMT_SYNC_SET_MNODE_STANDBY_RSP, mndTransProcessRsp);
  mndSetMsgHandle(pMnode, TDMT_SYNC_SET_VNODE_STANDBY_RSP, mndTransProcessRsp);
S
Shengliang Guan 已提交
61 62 63

  mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_MNODE, mndRetrieveMnodes);
  mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_MNODE, mndCancelGetNextMnode);
S
Shengliang Guan 已提交
64 65 66 67 68 69

  return sdbSetTable(pMnode->pSdb, table);
}

void mndCleanupMnode(SMnode *pMnode) {}

70 71
SMnodeObj *mndAcquireMnode(SMnode *pMnode, int32_t mnodeId) {
  SMnodeObj *pObj = sdbAcquire(pMnode->pSdb, SDB_MNODE, &mnodeId);
S
Shengliang Guan 已提交
72
  if (pObj == NULL && terrno == TSDB_CODE_SDB_OBJ_NOT_THERE) {
S
Shengliang Guan 已提交
73 74 75
    terrno = TSDB_CODE_MND_MNODE_NOT_EXIST;
  }
  return pObj;
S
Shengliang Guan 已提交
76 77
}

78
void mndReleaseMnode(SMnode *pMnode, SMnodeObj *pObj) {
S
Shengliang Guan 已提交
79
  SSdb *pSdb = pMnode->pSdb;
80
  sdbRelease(pMnode->pSdb, pObj);
S
Shengliang Guan 已提交
81 82
}

S
Shengliang Guan 已提交
83 84 85 86 87 88 89 90
static int32_t mndCreateDefaultMnode(SMnode *pMnode) {
  SMnodeObj mnodeObj = {0};
  mnodeObj.id = 1;
  mnodeObj.createdTime = taosGetTimestampMs();
  mnodeObj.updateTime = mnodeObj.createdTime;

  SSdbRaw *pRaw = mndMnodeActionEncode(&mnodeObj);
  if (pRaw == NULL) return -1;
S
Shengliang Guan 已提交
91
  (void)sdbSetRawStatus(pRaw, SDB_STATUS_READY);
S
Shengliang Guan 已提交
92

93
  mInfo("mnode:%d, will be created when deploying, raw:%p", mnodeObj.id, pRaw);
94

95
  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_GLOBAL, NULL, "create-mnode");
96
  if (pTrans == NULL) {
S
Shengliang Guan 已提交
97
    sdbFreeRaw(pRaw);
98 99 100
    mError("mnode:%d, failed to create since %s", mnodeObj.id, terrstr());
    return -1;
  }
101
  mInfo("trans:%d, used to create mnode:%d", pTrans->id, mnodeObj.id);
102 103 104 105 106 107

  if (mndTransAppendCommitlog(pTrans, pRaw) != 0) {
    mError("trans:%d, failed to append commit log since %s", pTrans->id, terrstr());
    mndTransDrop(pTrans);
    return -1;
  }
S
Shengliang Guan 已提交
108
  (void)sdbSetRawStatus(pRaw, SDB_STATUS_READY);
109 110 111 112 113 114 115 116 117

  if (mndTransPrepare(pMnode, pTrans) != 0) {
    mError("trans:%d, failed to prepare since %s", pTrans->id, terrstr());
    mndTransDrop(pTrans);
    return -1;
  }

  mndTransDrop(pTrans);
  return 0;
S
Shengliang Guan 已提交
118 119
}

S
Shengliang Guan 已提交
120
static SSdbRaw *mndMnodeActionEncode(SMnodeObj *pObj) {
121 122
  terrno = TSDB_CODE_OUT_OF_MEMORY;

123 124
  SSdbRaw *pRaw = sdbAllocRaw(SDB_MNODE, MNODE_VER_NUMBER, sizeof(SMnodeObj) + MNODE_RESERVE_SIZE);
  if (pRaw == NULL) goto _OVER;
S
Shengliang Guan 已提交
125 126

  int32_t dataPos = 0;
127 128 129 130
  SDB_SET_INT32(pRaw, dataPos, pObj->id, _OVER)
  SDB_SET_INT64(pRaw, dataPos, pObj->createdTime, _OVER)
  SDB_SET_INT64(pRaw, dataPos, pObj->updateTime, _OVER)
  SDB_SET_RESERVE(pRaw, dataPos, MNODE_RESERVE_SIZE, _OVER)
131 132 133

  terrno = 0;

134
_OVER:
135 136 137 138 139
  if (terrno != 0) {
    mError("mnode:%d, failed to encode to raw:%p since %s", pObj->id, pRaw, terrstr());
    sdbFreeRaw(pRaw);
    return NULL;
  }
S
Shengliang Guan 已提交
140

141
  mTrace("mnode:%d, encode to raw:%p, row:%p", pObj->id, pRaw, pObj);
S
Shengliang Guan 已提交
142 143 144 145
  return pRaw;
}

static SSdbRow *mndMnodeActionDecode(SSdbRaw *pRaw) {
146 147
  terrno = TSDB_CODE_OUT_OF_MEMORY;

S
Shengliang Guan 已提交
148 149 150
  int8_t sver = 0;
  if (sdbGetRawSoftVer(pRaw, &sver) != 0) return NULL;

151
  if (sver != MNODE_VER_NUMBER) {
S
Shengliang Guan 已提交
152
    terrno = TSDB_CODE_SDB_INVALID_DATA_VER;
153
    goto _OVER;
S
Shengliang Guan 已提交
154 155
  }

156
  SSdbRow *pRow = sdbAllocRow(sizeof(SMnodeObj));
157
  if (pRow == NULL) goto _OVER;
158

S
Shengliang Guan 已提交
159
  SMnodeObj *pObj = sdbGetRowObj(pRow);
160
  if (pObj == NULL) goto _OVER;
S
Shengliang Guan 已提交
161 162

  int32_t dataPos = 0;
163 164 165 166
  SDB_GET_INT32(pRaw, dataPos, &pObj->id, _OVER)
  SDB_GET_INT64(pRaw, dataPos, &pObj->createdTime, _OVER)
  SDB_GET_INT64(pRaw, dataPos, &pObj->updateTime, _OVER)
  SDB_GET_RESERVE(pRaw, dataPos, MNODE_RESERVE_SIZE, _OVER)
167 168 169

  terrno = 0;

170
_OVER:
171 172
  if (terrno != 0) {
    mError("mnode:%d, failed to decode from raw:%p since %s", pObj->id, pRaw, terrstr());
wafwerar's avatar
wafwerar 已提交
173
    taosMemoryFreeClear(pRow);
174 175
    return NULL;
  }
S
Shengliang Guan 已提交
176

177
  mTrace("mnode:%d, decode from raw:%p, row:%p", pObj->id, pRaw, pObj);
S
Shengliang Guan 已提交
178 179 180
  return pRow;
}

S
Shengliang Guan 已提交
181
static int32_t mndMnodeActionInsert(SSdb *pSdb, SMnodeObj *pObj) {
182
  mTrace("mnode:%d, perform insert action, row:%p", pObj->id, pObj);
S
Shengliang Guan 已提交
183 184
  pObj->pDnode = sdbAcquire(pSdb, SDB_DNODE, &pObj->id);
  if (pObj->pDnode == NULL) {
S
Shengliang Guan 已提交
185
    terrno = TSDB_CODE_MND_DNODE_NOT_EXIST;
S
Shengliang Guan 已提交
186
    mError("mnode:%d, failed to perform insert action since %s", pObj->id, terrstr());
S
Shengliang Guan 已提交
187 188 189
    return -1;
  }

190
  pObj->syncState = TAOS_SYNC_STATE_ERROR;
191
  mndReloadSyncConfig(pSdb->pMnode);
S
Shengliang Guan 已提交
192 193 194
  return 0;
}

S
Shengliang Guan 已提交
195
static int32_t mndMnodeActionDelete(SSdb *pSdb, SMnodeObj *pObj) {
196
  mTrace("mnode:%d, perform delete action, row:%p", pObj->id, pObj);
S
Shengliang Guan 已提交
197 198 199
  if (pObj->pDnode != NULL) {
    sdbRelease(pSdb, pObj->pDnode);
    pObj->pDnode = NULL;
S
Shengliang Guan 已提交
200 201 202 203 204
  }

  return 0;
}

S
Shengliang Guan 已提交
205
static int32_t mndMnodeActionUpdate(SSdb *pSdb, SMnodeObj *pOld, SMnodeObj *pNew) {
S
Shengliang Guan 已提交
206
  mTrace("mnode:%d, perform update action, old row:%p new row:%p", pOld->id, pOld, pNew);
S
Shengliang Guan 已提交
207
  pOld->updateTime = pNew->updateTime;
208 209
  mndReloadSyncConfig(pSdb->pMnode);

S
Shengliang Guan 已提交
210
  return 0;
S
Shengliang Guan 已提交
211 212 213 214 215
}

bool mndIsMnode(SMnode *pMnode, int32_t dnodeId) {
  SSdb *pSdb = pMnode->pSdb;

S
Shengliang Guan 已提交
216 217
  SMnodeObj *pObj = sdbAcquire(pSdb, SDB_MNODE, &dnodeId);
  if (pObj == NULL) {
S
Shengliang Guan 已提交
218 219 220
    return false;
  }

S
Shengliang Guan 已提交
221
  sdbRelease(pSdb, pObj);
S
Shengliang Guan 已提交
222
  return true;
S
Shengliang Guan 已提交
223 224
}

S
Shengliang Guan 已提交
225
void mndGetMnodeEpSet(SMnode *pMnode, SEpSet *pEpSet) {
226 227
  SSdb   *pSdb = pMnode->pSdb;
  int32_t totalMnodes = sdbGetSize(pSdb, SDB_MNODE);
S
Shengliang Guan 已提交
228 229 230 231
  if (totalMnodes == 0) {
    syncGetRetryEpSet(pMnode->syncMgmt.sync, pEpSet);
    return;
  }
S
Shengliang Guan 已提交
232

S
Shengliang Guan 已提交
233
  void *pIter = NULL;
S
Shengliang Guan 已提交
234
  while (1) {
S
Shengliang Guan 已提交
235 236
    SMnodeObj *pObj = NULL;
    pIter = sdbFetch(pSdb, SDB_MNODE, pIter, (void **)&pObj);
S
Shengliang Guan 已提交
237
    if (pIter == NULL) break;
238 239

    if (pObj->id == pMnode->selfDnodeId) {
240
      if (mndIsLeader(pMnode)) {
241
        pEpSet->inUse = pEpSet->numOfEps;
242 243
      } else {
        pEpSet->inUse = (pEpSet->numOfEps + 1) % totalMnodes;
244
      }
S
Shengliang Guan 已提交
245
    }
S
Shengliang Guan 已提交
246 247 248
    if (pObj->pDnode != NULL) {
      addEpIntoEpSet(pEpSet, pObj->pDnode->fqdn, pObj->pDnode->port);
    }
249
    sdbRelease(pSdb, pObj);
S
Shengliang Guan 已提交
250
  }
251 252 253 254

  if (pEpSet->numOfEps == 0) {
    syncGetRetryEpSet(pMnode->syncMgmt.sync, pEpSet);
  }
255 256 257 258

  if (pEpSet->inUse >= pEpSet->numOfEps) {
    pEpSet->inUse = 0;
  }
S
Shengliang Guan 已提交
259 260
}

261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284
static int32_t mndSetCreateMnodeRedoLogs(SMnode *pMnode, STrans *pTrans, SMnodeObj *pObj) {
  SSdbRaw *pRedoRaw = mndMnodeActionEncode(pObj);
  if (pRedoRaw == NULL) return -1;
  if (mndTransAppendRedolog(pTrans, pRedoRaw) != 0) return -1;
  if (sdbSetRawStatus(pRedoRaw, SDB_STATUS_CREATING) != 0) return -1;
  return 0;
}

static int32_t mndSetCreateMnodeUndoLogs(SMnode *pMnode, STrans *pTrans, SMnodeObj *pObj) {
  SSdbRaw *pUndoRaw = mndMnodeActionEncode(pObj);
  if (pUndoRaw == NULL) return -1;
  if (mndTransAppendUndolog(pTrans, pUndoRaw) != 0) return -1;
  if (sdbSetRawStatus(pUndoRaw, SDB_STATUS_DROPPED) != 0) return -1;
  return 0;
}

static int32_t mndSetCreateMnodeCommitLogs(SMnode *pMnode, STrans *pTrans, SMnodeObj *pObj) {
  SSdbRaw *pCommitRaw = mndMnodeActionEncode(pObj);
  if (pCommitRaw == NULL) return -1;
  if (mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) return -1;
  if (sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY) != 0) return -1;
  return 0;
}

285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326
static int32_t mndBuildCreateMnodeRedoAction(STrans *pTrans, SDCreateMnodeReq *pCreateReq, SEpSet *pCreateEpSet) {
  int32_t contLen = tSerializeSDCreateMnodeReq(NULL, 0, pCreateReq);
  void   *pReq = taosMemoryMalloc(contLen);
  tSerializeSDCreateMnodeReq(pReq, contLen, pCreateReq);

  STransAction action = {
      .epSet = *pCreateEpSet,
      .pCont = pReq,
      .contLen = contLen,
      .msgType = TDMT_DND_CREATE_MNODE,
      .acceptableCode = TSDB_CODE_NODE_ALREADY_DEPLOYED,
  };

  if (mndTransAppendRedoAction(pTrans, &action) != 0) {
    taosMemoryFree(pReq);
    return -1;
  }
  return 0;
}

static int32_t mndBuildAlterMnodeRedoAction(STrans *pTrans, SDCreateMnodeReq *pAlterReq, SEpSet *pAlterEpSet) {
  int32_t contLen = tSerializeSDCreateMnodeReq(NULL, 0, pAlterReq);
  void   *pReq = taosMemoryMalloc(contLen);
  tSerializeSDCreateMnodeReq(pReq, contLen, pAlterReq);

  STransAction action = {
      .epSet = *pAlterEpSet,
      .pCont = pReq,
      .contLen = contLen,
      .msgType = TDMT_MND_ALTER_MNODE,
      .acceptableCode = 0,
  };

  if (mndTransAppendRedoAction(pTrans, &action) != 0) {
    taosMemoryFree(pReq);
    return -1;
  }

  return 0;
}

static int32_t mndBuildDropMnodeRedoAction(STrans *pTrans, SDDropMnodeReq *pDropReq, SEpSet *pDroprEpSet) {
S
Shengliang Guan 已提交
327
  int32_t contLen = tSerializeSCreateDropMQSNodeReq(NULL, 0, pDropReq);
328
  void   *pReq = taosMemoryMalloc(contLen);
S
Shengliang Guan 已提交
329
  tSerializeSCreateDropMQSNodeReq(pReq, contLen, pDropReq);
330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345

  STransAction action = {
      .epSet = *pDroprEpSet,
      .pCont = pReq,
      .contLen = contLen,
      .msgType = TDMT_DND_DROP_MNODE,
      .acceptableCode = TSDB_CODE_NODE_NOT_DEPLOYED,
  };

  if (mndTransAppendRedoAction(pTrans, &action) != 0) {
    taosMemoryFree(pReq);
    return -1;
  }
  return 0;
}

S
Shengliang Guan 已提交
346
static int32_t mndSetCreateMnodeRedoActions(SMnode *pMnode, STrans *pTrans, SDnodeObj *pDnode, SMnodeObj *pObj) {
S
Shengliang Guan 已提交
347 348 349
  SSdb            *pSdb = pMnode->pSdb;
  void            *pIter = NULL;
  int32_t          numOfReplicas = 0;
S
Shengliang Guan 已提交
350
  SDCreateMnodeReq createReq = {0};
S
Shengliang Guan 已提交
351 352
  SEpSet           createEpset = {0};

S
Shengliang Guan 已提交
353 354 355 356
  while (1) {
    SMnodeObj *pMObj = NULL;
    pIter = sdbFetch(pSdb, SDB_MNODE, pIter, (void **)&pMObj);
    if (pIter == NULL) break;
357

358 359 360
    createReq.replicas[numOfReplicas].id = pMObj->id;
    createReq.replicas[numOfReplicas].port = pMObj->pDnode->port;
    memcpy(createReq.replicas[numOfReplicas].fqdn, pMObj->pDnode->fqdn, TSDB_FQDN_LEN);
S
Shengliang Guan 已提交
361 362

    numOfReplicas++;
S
Shengliang Guan 已提交
363 364 365
    sdbRelease(pSdb, pMObj);
  }

366 367 368 369
  createReq.replica = numOfReplicas + 1;
  createReq.replicas[numOfReplicas].id = pDnode->id;
  createReq.replicas[numOfReplicas].port = pDnode->port;
  memcpy(createReq.replicas[numOfReplicas].fqdn, pDnode->fqdn, TSDB_FQDN_LEN);
S
Shengliang Guan 已提交
370

371
  createEpset.inUse = 0;
S
Shengliang Guan 已提交
372 373 374
  createEpset.numOfEps = 1;
  createEpset.eps[0].port = pDnode->port;
  memcpy(createEpset.eps[0].fqdn, pDnode->fqdn, TSDB_FQDN_LEN);
S
Shengliang Guan 已提交
375

376
  if (mndBuildCreateMnodeRedoAction(pTrans, &createReq, &createEpset) != 0) return -1;
377 378 379 380

  return 0;
}

S
Shengliang Guan 已提交
381
static int32_t mndCreateMnode(SMnode *pMnode, SRpcMsg *pReq, SDnodeObj *pDnode, SMCreateMnodeReq *pCreate) {
S
Shengliang Guan 已提交
382 383
  int32_t code = -1;

S
Shengliang Guan 已提交
384
  SMnodeObj mnodeObj = {0};
S
Shengliang Guan 已提交
385
  mnodeObj.id = pDnode->id;
S
Shengliang Guan 已提交
386 387 388
  mnodeObj.createdTime = taosGetTimestampMs();
  mnodeObj.updateTime = mnodeObj.createdTime;

389
  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_GLOBAL, pReq, "create-mnode");
390
  if (pTrans == NULL) goto _OVER;
391
  mndTransSetSerial(pTrans);
392
  mInfo("trans:%d, used to create mnode:%d", pTrans->id, pCreate->dnodeId);
S
Shengliang Guan 已提交
393

394
  if (mndSetCreateMnodeRedoActions(pMnode, pTrans, pDnode, &mnodeObj) != 0) goto _OVER;
395 396 397
  if (mndSetCreateMnodeRedoLogs(pMnode, pTrans, &mnodeObj) != 0) goto _OVER;
  if (mndSetCreateMnodeCommitLogs(pMnode, pTrans, &mnodeObj) != 0) goto _OVER;
  if (mndTransPrepare(pMnode, pTrans) != 0) goto _OVER;
S
Shengliang Guan 已提交
398

399 400
  code = 0;

401
_OVER:
S
Shengliang Guan 已提交
402
  mndTransDrop(pTrans);
403
  return code;
S
Shengliang Guan 已提交
404 405
}

S
Shengliang Guan 已提交
406 407
static int32_t mndProcessCreateMnodeReq(SRpcMsg *pReq) {
  SMnode          *pMnode = pReq->info.node;
S
Shengliang Guan 已提交
408 409 410 411 412
  int32_t          code = -1;
  SMnodeObj       *pObj = NULL;
  SDnodeObj       *pDnode = NULL;
  SMCreateMnodeReq createReq = {0};

S
Shengliang Guan 已提交
413
  if (tDeserializeSCreateDropMQSNodeReq(pReq->pCont, pReq->contLen, &createReq) != 0) {
S
Shengliang Guan 已提交
414
    terrno = TSDB_CODE_INVALID_MSG;
415
    goto _OVER;
S
Shengliang Guan 已提交
416
  }
S
Shengliang Guan 已提交
417

418
  mInfo("mnode:%d, start to create", createReq.dnodeId);
419
  if (mndCheckOperPrivilege(pMnode, pReq->info.conn.user, MND_OPER_CREATE_MNODE) != 0) {
S
Shengliang Guan 已提交
420 421
    goto _OVER;
  }
S
Shengliang Guan 已提交
422

S
Shengliang Guan 已提交
423
  pObj = mndAcquireMnode(pMnode, createReq.dnodeId);
S
Shengliang Guan 已提交
424
  if (pObj != NULL) {
S
Shengliang Guan 已提交
425
    terrno = TSDB_CODE_MND_MNODE_ALREADY_EXIST;
426
    goto _OVER;
S
Shengliang Guan 已提交
427
  } else if (terrno != TSDB_CODE_MND_MNODE_NOT_EXIST) {
428
    goto _OVER;
S
Shengliang Guan 已提交
429 430
  }

S
Shengliang Guan 已提交
431
  pDnode = mndAcquireDnode(pMnode, createReq.dnodeId);
S
Shengliang Guan 已提交
432 433
  if (pDnode == NULL) {
    terrno = TSDB_CODE_MND_DNODE_NOT_EXIST;
434
    goto _OVER;
S
Shengliang Guan 已提交
435 436
  }

437 438 439 440 441
  if (sdbGetSize(pMnode->pSdb, SDB_MNODE) >= 3) {
    terrno = TSDB_CODE_MND_TOO_MANY_MNODES;
    goto _OVER;
  }

S
Shengliang Guan 已提交
442
  if (!mndIsDnodeOnline(pDnode, taosGetTimestampMs())) {
443 444 445 446
    terrno = TSDB_CODE_NODE_OFFLINE;
    goto _OVER;
  }

S
Shengliang Guan 已提交
447
  code = mndCreateMnode(pMnode, pReq, pDnode, &createReq);
S
Shengliang Guan 已提交
448
  if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
S
Shengliang Guan 已提交
449

450
_OVER:
S
Shengliang Guan 已提交
451
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {
S
Shengliang Guan 已提交
452
    mError("mnode:%d, failed to create since %s", createReq.dnodeId, terrstr());
S
Shengliang Guan 已提交
453 454
  }

S
Shengliang Guan 已提交
455 456 457 458
  mndReleaseMnode(pMnode, pObj);
  mndReleaseDnode(pMnode, pDnode);

  return code;
S
Shengliang Guan 已提交
459 460
}

461 462 463 464 465 466 467 468 469 470 471 472 473 474 475 476
static int32_t mndSetDropMnodeRedoLogs(SMnode *pMnode, STrans *pTrans, SMnodeObj *pObj) {
  SSdbRaw *pRedoRaw = mndMnodeActionEncode(pObj);
  if (pRedoRaw == NULL) return -1;
  if (mndTransAppendRedolog(pTrans, pRedoRaw) != 0) return -1;
  if (sdbSetRawStatus(pRedoRaw, SDB_STATUS_DROPPING) != 0) return -1;
  return 0;
}

static int32_t mndSetDropMnodeCommitLogs(SMnode *pMnode, STrans *pTrans, SMnodeObj *pObj) {
  SSdbRaw *pCommitRaw = mndMnodeActionEncode(pObj);
  if (pCommitRaw == NULL) return -1;
  if (mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) return -1;
  if (sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED) != 0) return -1;
  return 0;
}

S
Shengliang Guan 已提交
477
static int32_t mndSetDropMnodeRedoActions(SMnode *pMnode, STrans *pTrans, SDnodeObj *pDnode, SMnodeObj *pObj) {
478 479 480 481
  SSdb          *pSdb = pMnode->pSdb;
  void          *pIter = NULL;
  SDDropMnodeReq dropReq = {0};
  SEpSet         dropEpSet = {0};
S
Shengliang Guan 已提交
482

S
Shengliang Guan 已提交
483 484 485 486
  dropReq.dnodeId = pDnode->id;
  dropEpSet.numOfEps = 1;
  dropEpSet.eps[0].port = pDnode->port;
  memcpy(dropEpSet.eps[0].fqdn, pDnode->fqdn, TSDB_FQDN_LEN);
S
Shengliang Guan 已提交
487

488 489 490 491 492 493 494 495 496 497 498
  int32_t totalMnodes = sdbGetSize(pSdb, SDB_MNODE);
  if (totalMnodes == 2) {
    mInfo("vgId:1, has %d mnodes, exec redo log first", totalMnodes);
    if (mndSetDropMnodeRedoLogs(pMnode, pTrans, pObj) != 0) return -1;
    if (mndBuildDropMnodeRedoAction(pTrans, &dropReq, &dropEpSet) != 0) return -1;
  } else if (totalMnodes == 3) {
    mInfo("vgId:1, has %d mnodes, exec redo action first", totalMnodes);
    if (mndBuildDropMnodeRedoAction(pTrans, &dropReq, &dropEpSet) != 0) return -1;
    if (mndSetDropMnodeRedoLogs(pMnode, pTrans, pObj) != 0) return -1;
  } else {
    return -1;
499 500 501 502 503
  }

  return 0;
}

S
Shengliang Guan 已提交
504
int32_t mndSetDropMnodeInfoToTrans(SMnode *pMnode, STrans *pTrans, SMnodeObj *pObj) {
505
  if (pObj == NULL) return 0;
S
Shengliang Guan 已提交
506
  if (mndSetDropMnodeRedoActions(pMnode, pTrans, pObj->pDnode, pObj) != 0) return -1;
507
  if (mndSetDropMnodeCommitLogs(pMnode, pTrans, pObj) != 0) return -1;
S
Shengliang Guan 已提交
508 509 510
  return 0;
}

S
Shengliang Guan 已提交
511
static int32_t mndDropMnode(SMnode *pMnode, SRpcMsg *pReq, SMnodeObj *pObj) {
512
  int32_t code = -1;
S
Shengliang Guan 已提交
513
  STrans *pTrans = NULL;
S
Shengliang Guan 已提交
514

515
  pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_GLOBAL, pReq, "drop-mnode");
516
  if (pTrans == NULL) goto _OVER;
S
Shengliang Guan 已提交
517
  mndTransSetSerial(pTrans);
518
  mInfo("trans:%d, used to drop mnode:%d", pTrans->id, pObj->id);
S
Shengliang Guan 已提交
519

S
Shengliang Guan 已提交
520
  if (mndSetDropMnodeInfoToTrans(pMnode, pTrans, pObj) != 0) goto _OVER;
521
  if (mndTransPrepare(pMnode, pTrans) != 0) goto _OVER;
S
Shengliang Guan 已提交
522

523 524
  code = 0;

525
_OVER:
S
Shengliang Guan 已提交
526
  mndTransDrop(pTrans);
527
  return code;
S
Shengliang Guan 已提交
528 529
}

S
Shengliang Guan 已提交
530 531
static int32_t mndProcessDropMnodeReq(SRpcMsg *pReq) {
  SMnode        *pMnode = pReq->info.node;
S
Shengliang Guan 已提交
532 533 534 535
  int32_t        code = -1;
  SMnodeObj     *pObj = NULL;
  SMDropMnodeReq dropReq = {0};

S
Shengliang Guan 已提交
536
  if (tDeserializeSCreateDropMQSNodeReq(pReq->pCont, pReq->contLen, &dropReq) != 0) {
S
Shengliang Guan 已提交
537
    terrno = TSDB_CODE_INVALID_MSG;
538
    goto _OVER;
S
Shengliang Guan 已提交
539
  }
S
Shengliang Guan 已提交
540

541
  mInfo("mnode:%d, start to drop", dropReq.dnodeId);
542
  if (mndCheckOperPrivilege(pMnode, pReq->info.conn.user, MND_OPER_DROP_MNODE) != 0) {
S
Shengliang Guan 已提交
543 544
    goto _OVER;
  }
S
Shengliang Guan 已提交
545

S
Shengliang Guan 已提交
546
  if (dropReq.dnodeId <= 0) {
547 548
    terrno = TSDB_CODE_INVALID_MSG;
    goto _OVER;
S
Shengliang Guan 已提交
549 550
  }

S
Shengliang Guan 已提交
551
  pObj = mndAcquireMnode(pMnode, dropReq.dnodeId);
S
Shengliang Guan 已提交
552
  if (pObj == NULL) {
553 554 555
    goto _OVER;
  }

556
  if (pMnode->selfDnodeId == dropReq.dnodeId) {
S
Shengliang Guan 已提交
557
    terrno = TSDB_CODE_MND_CANT_DROP_LEADER;
558 559 560 561 562 563
    goto _OVER;
  }

  if (sdbGetSize(pMnode->pSdb, SDB_MNODE) <= 1) {
    terrno = TSDB_CODE_MND_TOO_FEW_MNODES;
    goto _OVER;
S
Shengliang Guan 已提交
564 565
  }

566 567 568 569 570
  if (!mndIsDnodeOnline(pObj->pDnode, taosGetTimestampMs())) {
    terrno = TSDB_CODE_NODE_OFFLINE;
    goto _OVER;
  }

S
Shengliang Guan 已提交
571
  code = mndDropMnode(pMnode, pReq, pObj);
S
Shengliang Guan 已提交
572
  if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;
S
Shengliang Guan 已提交
573

574
_OVER:
S
Shengliang Guan 已提交
575
  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {
S
Shengliang Guan 已提交
576 577 578 579 580
    mError("mnode:%d, failed to drop since %s", dropReq.dnodeId, terrstr());
  }

  mndReleaseMnode(pMnode, pObj);
  return code;
S
Shengliang Guan 已提交
581 582
}

S
Shengliang Guan 已提交
583 584
static int32_t mndRetrieveMnodes(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {
  SMnode    *pMnode = pReq->info.node;
S
Shengliang Guan 已提交
585 586 587
  SSdb      *pSdb = pMnode->pSdb;
  int32_t    numOfRows = 0;
  int32_t    cols = 0;
S
Shengliang Guan 已提交
588
  SMnodeObj *pObj = NULL;
589
  ESdbStatus objStatus = 0;
S
Shengliang Guan 已提交
590
  char      *pWrite;
591
  int64_t    curMs = taosGetTimestampMs();
S
Shengliang Guan 已提交
592 593

  while (numOfRows < rows) {
594
    pShow->pIter = sdbFetchAll(pSdb, SDB_MNODE, pShow->pIter, (void **)&pObj, &objStatus, true);
S
Shengliang Guan 已提交
595 596 597
    if (pShow->pIter == NULL) break;

    cols = 0;
S
Shengliang Guan 已提交
598 599
    SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    colDataAppend(pColInfo, numOfRows, (const char *)&pObj->id, false);
S
Shengliang Guan 已提交
600

601
    char b1[TSDB_EP_LEN + VARSTR_HEADER_SIZE] = {0};
S
Shengliang Guan 已提交
602
    STR_WITH_MAXSIZE_TO_VARSTR(b1, pObj->pDnode->ep, TSDB_EP_LEN + VARSTR_HEADER_SIZE);
S
Shengliang Guan 已提交
603

604 605
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    colDataAppend(pColInfo, numOfRows, b1, false);
S
Shengliang Guan 已提交
606

607
    char role[20] = "offline";
608
    if (pObj->id == pMnode->selfDnodeId) {
609
      snprintf(role, sizeof(role), "%s%s", syncStr(TAOS_SYNC_STATE_LEADER), pMnode->restored ? "" : "*");
610
    }
S
Shengliang Guan 已提交
611
    if (mndIsDnodeOnline(pObj->pDnode, curMs)) {
612 613 614
      tstrncpy(role, syncStr(pObj->syncState), sizeof(role));
      if (pObj->syncState == TAOS_SYNC_STATE_LEADER && pObj->id != pMnode->selfDnodeId) {
        tstrncpy(role, syncStr(TAOS_SYNC_STATE_ERROR), sizeof(role));
S
Shengliang Guan 已提交
615 616
        mError("mnode:%d, is leader too", pObj->id);
      }
617 618
    }
    char b2[12 + VARSTR_HEADER_SIZE] = {0};
619
    STR_WITH_MAXSIZE_TO_VARSTR(b2, role, pShow->pMeta->pSchemas[cols].bytes);
620
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
S
Shengliang Guan 已提交
621
    colDataAppend(pColInfo, numOfRows, (const char *)b2, false);
S
Shengliang Guan 已提交
622

623 624 625
    const char *status = "ready";
    if (objStatus == SDB_STATUS_CREATING) status = "creating";
    if (objStatus == SDB_STATUS_DROPPING) status = "dropping";
626 627 628 629 630
    char b3[9 + VARSTR_HEADER_SIZE] = {0};
    STR_WITH_MAXSIZE_TO_VARSTR(b3, status, pShow->pMeta->pSchemas[cols].bytes);
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    colDataAppend(pColInfo, numOfRows, (const char *)b3, false);

631 632
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    colDataAppend(pColInfo, numOfRows, (const char *)&pObj->createdTime, false);
S
Shengliang Guan 已提交
633 634

    numOfRows++;
S
Shengliang Guan 已提交
635
    sdbRelease(pSdb, pObj);
S
Shengliang Guan 已提交
636 637
  }

638
  pShow->numOfRows += numOfRows;
S
Shengliang Guan 已提交
639 640 641 642 643 644 645 646

  return numOfRows;
}

static void mndCancelGetNextMnode(SMnode *pMnode, void *pIter) {
  SSdb *pSdb = pMnode->pSdb;
  sdbCancelFetch(pSdb, pIter);
}
S
Shengliang Guan 已提交
647 648

static int32_t mndProcessAlterMnodeReq(SRpcMsg *pReq) {
649 650 651
#if 1
  return 0;
#else
S
Shengliang Guan 已提交
652
  SMnode         *pMnode = pReq->info.node;
S
Shengliang Guan 已提交
653 654 655 656 657 658 659
  SDAlterMnodeReq alterReq = {0};

  if (tDeserializeSDCreateMnodeReq(pReq->pCont, pReq->contLen, &alterReq) != 0) {
    terrno = TSDB_CODE_INVALID_MSG;
    return -1;
  }

660 661 662 663 664 665 666 667 668 669 670 671 672 673 674 675 676 677
  SMnodeOpt option = {.deploy = true, .numOfReplicas = alterReq.replica, .selfIndex = -1};
  memcpy(option.replicas, alterReq.replicas, sizeof(alterReq.replicas));
  for (int32_t i = 0; i < option.numOfReplicas; ++i) {
    if (alterReq.replicas[i].id == pMnode->selfDnodeId) {
      option.selfIndex = i;
    }
  }

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

  if (mndWriteFile(pMnode->path, &option) != 0) {
    mError("failed to write mnode file since %s", terrstr());
    return -1;
  }

S
Shengliang Guan 已提交
678 679 680 681 682
  SSyncCfg cfg = {.replicaNum = alterReq.replica, .myIndex = -1};
  for (int32_t i = 0; i < alterReq.replica; ++i) {
    SNodeInfo *pNode = &cfg.nodeInfo[i];
    tstrncpy(pNode->nodeFqdn, alterReq.replicas[i].fqdn, sizeof(pNode->nodeFqdn));
    pNode->nodePort = alterReq.replicas[i].port;
683 684 685
    if (alterReq.replicas[i].id == pMnode->selfDnodeId) {
      cfg.myIndex = i;
    }
S
Shengliang Guan 已提交
686 687 688 689 690 691
  }

  if (cfg.myIndex == -1) {
    mError("failed to alter mnode since myindex is -1");
    return -1;
  } else {
692
    mInfo("start to alter mnode sync, replica:%d myIndex:%d", cfg.replicaNum, cfg.myIndex);
S
Shengliang Guan 已提交
693 694 695 696 697 698
    for (int32_t i = 0; i < alterReq.replica; ++i) {
      SNodeInfo *pNode = &cfg.nodeInfo[i];
      mInfo("index:%d, fqdn:%s port:%d", i, pNode->nodeFqdn, pNode->nodePort);
    }
  }

699
  int32_t code = syncReconfig(pMnode->syncMgmt.sync, &cfg);
S
Shengliang Guan 已提交
700
  if (code != 0) {
701
    mError("failed to sync reconfig since %s", terrstr());
S
Shengliang Guan 已提交
702
  } else {
703 704 705 706 707 708 709 710 711 712 713 714
    mInfo("alter mnode sync success");
  }

  return code;
#endif
}

static void mndReloadSyncConfig(SMnode *pMnode) {
  SSdb      *pSdb = pMnode->pSdb;
  SMnodeObj *pObj = NULL;
  ESdbStatus objStatus = 0;
  void      *pIter = NULL;
715 716
  int32_t    updatingMnodes = 0;
  int32_t    readyMnodes = 0;
717 718 719 720 721 722 723
  SSyncCfg   cfg = {.myIndex = -1};

  while (1) {
    pIter = sdbFetchAll(pSdb, SDB_MNODE, pIter, (void **)&pObj, &objStatus, false);
    if (pIter == NULL) break;
    if (objStatus == SDB_STATUS_CREATING || objStatus == SDB_STATUS_DROPPING) {
      mInfo("vgId:1, has updating mnode:%d, status:%s", pObj->id, sdbStatusName(objStatus));
724 725 726 727 728
      updatingMnodes++;
    }
    if (objStatus == SDB_STATUS_READY) {
      mInfo("vgId:1, has ready mnode:%d, status:%s", pObj->id, sdbStatusName(objStatus));
      readyMnodes++;
729 730 731 732 733 734 735 736 737 738 739 740 741 742 743
    }

    if (objStatus == SDB_STATUS_READY || objStatus == SDB_STATUS_CREATING) {
      SNodeInfo *pNode = &cfg.nodeInfo[cfg.replicaNum];
      tstrncpy(pNode->nodeFqdn, pObj->pDnode->fqdn, sizeof(pNode->nodeFqdn));
      pNode->nodePort = pObj->pDnode->port;
      if (pObj->pDnode->id == pMnode->selfDnodeId) {
        cfg.myIndex = cfg.replicaNum;
      }
      cfg.replicaNum++;
    }

    sdbReleaseLock(pSdb, pObj, false);
  }

744 745
  if (readyMnodes <= 0 || updatingMnodes <= 0) {
    mInfo("vgId:1, mnode sync not reconfig since readyMnodes:%d updatingMnodes:%d", readyMnodes, updatingMnodes);
746 747
    return;
  }
748
    // ASSERT(0);
749

750
  if (cfg.myIndex == -1) {
751
#if 1
752
    mInfo("vgId:1, mnode sync not reconfig since selfIndex is -1");
753 754 755 756 757
#else
    // cannot reconfig because the leader may fail to elect after reboot
    mInfo("vgId:1, mnode sync not reconfig since selfIndex is -1, do sync stop oper");
    syncStop(pMnode->syncMgmt.sync);
#endif
758 759 760
    return;
  }

761 762
  if (updatingMnodes > 0) {
    mInfo("vgId:1, mnode sync reconfig, replica:%d myIndex:%d", cfg.replicaNum, cfg.myIndex);
763 764 765 766 767 768 769 770 771 772 773
    for (int32_t i = 0; i < cfg.replicaNum; ++i) {
      SNodeInfo *pNode = &cfg.nodeInfo[i];
      mInfo("vgId:1, index:%d, fqdn:%s port:%d", i, pNode->nodeFqdn, pNode->nodePort);
    }

    int32_t code = syncReconfig(pMnode->syncMgmt.sync, &cfg);
    if (code != 0) {
      mError("vgId:1, failed to reconfig mnode sync since %s", terrstr());
    } else {
      mInfo("vgId:1, reconfig mnode sync success");
    }
S
Shengliang Guan 已提交
774 775
  }
}