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

#define _DEFAULT_SOURCE
S
Shengliang Guan 已提交
17 18
#include "mndAcct.h"
#include "mndAuth.h"
S
Shengliang Guan 已提交
19
#include "mndBnode.h"
S
Shengliang Guan 已提交
20
#include "mndCluster.h"
L
Liu Jicong 已提交
21
#include "mndConsumer.h"
S
Shengliang Guan 已提交
22 23 24
#include "mndDb.h"
#include "mndDnode.h"
#include "mndFunc.h"
L
Liu Jicong 已提交
25
#include "mndGrant.h"
S
monitor  
Shengliang Guan 已提交
26
#include "mndInfoSchema.h"
S
Shengliang Guan 已提交
27
#include "mndMnode.h"
L
Liu Jicong 已提交
28
#include "mndOffset.h"
L
Liu Jicong 已提交
29
#include "mndPerfSchema.h"
S
Shengliang Guan 已提交
30
#include "mndProfile.h"
S
Shengliang Guan 已提交
31
#include "mndQnode.h"
L
Liu Jicong 已提交
32
#include "mndQuery.h"
S
Shengliang Guan 已提交
33
#include "mndShow.h"
S
sma  
Shengliang Guan 已提交
34
#include "mndSma.h"
S
Shengliang Guan 已提交
35
#include "mndSnode.h"
S
Shengliang Guan 已提交
36
#include "mndStb.h"
L
Liu Jicong 已提交
37
#include "mndStream.h"
L
Liu Jicong 已提交
38
#include "mndSubscribe.h"
S
Shengliang Guan 已提交
39 40
#include "mndSync.h"
#include "mndTelem.h"
L
Liu Jicong 已提交
41
#include "mndTopic.h"
S
Shengliang Guan 已提交
42 43 44
#include "mndTrans.h"
#include "mndUser.h"
#include "mndVgroup.h"
S
Shengliang Guan 已提交
45

S
Shengliang Guan 已提交
46 47 48 49 50 51 52 53 54 55 56 57 58
static void *mndBuildTimerMsg(int32_t *pContLen) {
  SMTimerReq timerReq = {0};

  int32_t contLen = tSerializeSMTimerMsg(NULL, 0, &timerReq);
  if (contLen <= 0) return NULL;
  void *pReq = rpcMallocCont(contLen);
  if (pReq == NULL) return NULL;

  tSerializeSMTimerMsg(pReq, contLen, &timerReq);
  *pContLen = contLen;
  return pReq;
}

S
Shengliang Guan 已提交
59 60 61 62 63
static void mndPullupTrans(SMnode *pMnode) {
  int32_t contLen = 0;
  void   *pReq = mndBuildTimerMsg(&contLen);
  SRpcMsg rpcMsg = {.msgType = TDMT_MND_TRANS_TIMER, .pCont = pReq, .contLen = contLen};
  tmsgPutToQueue(&pMnode->msgCb, WRITE_QUEUE, &rpcMsg);
S
Shengliang Guan 已提交
64 65
}

S
Shengliang Guan 已提交
66 67 68
static void mndCalMqRebalance(SMnode *pMnode) {
  int32_t contLen = 0;
  void   *pReq = mndBuildTimerMsg(&contLen);
69
  SRpcMsg rpcMsg = {.msgType = TDMT_MND_MQ_TIMER, .pCont = pReq, .contLen = contLen};
S
Shengliang Guan 已提交
70 71
  tmsgPutToQueue(&pMnode->msgCb, READ_QUEUE, &rpcMsg);
}
L
Liu Jicong 已提交
72

S
Shengliang Guan 已提交
73 74 75 76
static void mndPullupTelem(SMnode *pMnode) {
  int32_t contLen = 0;
  void   *pReq = mndBuildTimerMsg(&contLen);
  SRpcMsg rpcMsg = {.msgType = TDMT_MND_TELEM_TIMER, .pCont = pReq, .contLen = contLen};
77
  tmsgPutToQueue(&pMnode->msgCb, READ_QUEUE, &rpcMsg);
S
Shengliang Guan 已提交
78 79
}

S
Shengliang Guan 已提交
80
static void *mndThreadFp(void *param) {
S
Shengliang Guan 已提交
81
  SMnode *pMnode = param;
S
Shengliang Guan 已提交
82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101
  int64_t lastTime = 0;
  setThreadName("mnode-timer");

  while (1) {
    lastTime++;
    taosMsleep(100);
    if (pMnode->stopped) break;
    if (!mndIsMaster(pMnode)) continue;

    if (lastTime % (tsTransPullupInterval * 10) == 0) {
      mndPullupTrans(pMnode);
    }

    if (lastTime % (tsMqRebalanceInterval * 10) == 0) {
      mndCalMqRebalance(pMnode);
    }

    if (lastTime % (tsTelemInterval * 10) == 0) {
      mndPullupTelem(pMnode);
    }
S
Shengliang Guan 已提交
102 103
  }

S
Shengliang Guan 已提交
104
  return NULL;
L
Liu Jicong 已提交
105 106
}

107
static int32_t mndInitTimer(SMnode *pMnode) {
S
Shengliang Guan 已提交
108 109 110 111 112
  TdThreadAttr thAttr;
  taosThreadAttrInit(&thAttr);
  taosThreadAttrSetDetachState(&thAttr, PTHREAD_CREATE_JOINABLE);
  if (taosThreadCreate(&pMnode->thread, &thAttr, mndThreadFp, pMnode) != 0) {
    mError("failed to create timer thread since %s", strerror(errno));
S
shm  
Shengliang Guan 已提交
113 114
    return -1;
  }
L
Liu Jicong 已提交
115

S
Shengliang Guan 已提交
116 117
  taosThreadAttrDestroy(&thAttr);
  tmsgReportStartup("mnode-timer", "initialized");
S
Shengliang Guan 已提交
118 119 120
  return 0;
}

121
static void mndCleanupTimer(SMnode *pMnode) {
S
Shengliang Guan 已提交
122 123 124
  pMnode->stopped = true;
  if (taosCheckPthreadValid(pMnode->thread)) {
    taosThreadJoin(pMnode->thread, NULL);
125
    taosThreadClear(&pMnode->thread);
S
Shengliang Guan 已提交
126 127 128
  }
}

S
Shengliang Guan 已提交
129
static int32_t mndCreateDir(SMnode *pMnode, const char *path) {
130 131 132
  pMnode->path = strdup(path);
  if (pMnode->path == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
S
Shengliang Guan 已提交
133
    return -1;
134 135 136 137
  }

  if (taosMkDir(pMnode->path) != 0) {
    terrno = TAOS_SYSTEM_ERROR(errno);
S
Shengliang Guan 已提交
138
    return -1;
139
  }
140 141

  return 0;
142
}
S
Shengliang Guan 已提交
143

144 145 146
static int32_t mndInitSdb(SMnode *pMnode) {
  SSdbOpt opt = {0};
  opt.path = pMnode->path;
S
Shengliang Guan 已提交
147
  opt.pMnode = pMnode;
S
Shengliang Guan 已提交
148

S
Shengliang Guan 已提交
149
  pMnode->pSdb = sdbInit(&opt);
150
  if (pMnode->pSdb == NULL) {
S
Shengliang Guan 已提交
151 152 153 154 155 156
    return -1;
  }

  return 0;
}

157
static int32_t mndDeploySdb(SMnode *pMnode) { return sdbDeploy(pMnode->pSdb); }
S
Shengliang Guan 已提交
158
static int32_t mndReadSdb(SMnode *pMnode) { return sdbReadFile(pMnode->pSdb); }
159 160 161

static void mndCleanupSdb(SMnode *pMnode) {
  if (pMnode->pSdb) {
S
Shengliang Guan 已提交
162
    sdbCleanup(pMnode->pSdb);
163 164 165 166
    pMnode->pSdb = NULL;
  }
}

167 168 169 170 171
static int32_t mndAllocStep(SMnode *pMnode, char *name, MndInitFp initFp, MndCleanupFp cleanupFp) {
  SMnodeStep step = {0};
  step.name = name;
  step.initFp = initFp;
  step.cleanupFp = cleanupFp;
S
Shengliang Guan 已提交
172
  if (taosArrayPush(pMnode->pSteps, &step) == NULL) {
173 174 175 176
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return -1;
  }

S
Shengliang Guan 已提交
177 178 179
  return 0;
}

S
shm  
Shengliang Guan 已提交
180
static int32_t mndInitSteps(SMnode *pMnode, bool deploy) {
S
Shengliang Guan 已提交
181 182 183 184
  if (mndAllocStep(pMnode, "mnode-sdb", mndInitSdb, mndCleanupSdb) != 0) return -1;
  if (mndAllocStep(pMnode, "mnode-trans", mndInitTrans, mndCleanupTrans) != 0) return -1;
  if (mndAllocStep(pMnode, "mnode-cluster", mndInitCluster, mndCleanupCluster) != 0) return -1;
  if (mndAllocStep(pMnode, "mnode-mnode", mndInitMnode, mndCleanupMnode) != 0) return -1;
S
Shengliang Guan 已提交
185 186 187
  if (mndAllocStep(pMnode, "mnode-qnode", mndInitQnode, mndCleanupQnode) != 0) return -1;
  if (mndAllocStep(pMnode, "mnode-qnode", mndInitSnode, mndCleanupSnode) != 0) return -1;
  if (mndAllocStep(pMnode, "mnode-qnode", mndInitBnode, mndCleanupBnode) != 0) return -1;
S
Shengliang Guan 已提交
188
  if (mndAllocStep(pMnode, "mnode-dnode", mndInitDnode, mndCleanupDnode) != 0) return -1;
S
Shengliang Guan 已提交
189
  if (mndAllocStep(pMnode, "mnode-user", mndInitUser, mndCleanupUser) != 0) return -1;
wafwerar's avatar
wafwerar 已提交
190
  if (mndAllocStep(pMnode, "mnode-grant", mndInitGrant, mndCleanupGrant) != 0) return -1;
S
Shengliang Guan 已提交
191 192
  if (mndAllocStep(pMnode, "mnode-auth", mndInitAuth, mndCleanupAuth) != 0) return -1;
  if (mndAllocStep(pMnode, "mnode-acct", mndInitAcct, mndCleanupAcct) != 0) return -1;
L
Liu Jicong 已提交
193
  if (mndAllocStep(pMnode, "mnode-stream", mndInitStream, mndCleanupStream) != 0) return -1;
S
Shengliang Guan 已提交
194
  if (mndAllocStep(pMnode, "mnode-topic", mndInitTopic, mndCleanupTopic) != 0) return -1;
L
Liu Jicong 已提交
195 196
  if (mndAllocStep(pMnode, "mnode-consumer", mndInitConsumer, mndCleanupConsumer) != 0) return -1;
  if (mndAllocStep(pMnode, "mnode-subscribe", mndInitSubscribe, mndCleanupSubscribe) != 0) return -1;
L
Liu Jicong 已提交
197
  if (mndAllocStep(pMnode, "mnode-offset", mndInitOffset, mndCleanupOffset) != 0) return -1;
S
Shengliang Guan 已提交
198
  if (mndAllocStep(pMnode, "mnode-vgroup", mndInitVgroup, mndCleanupVgroup) != 0) return -1;
S
Shengliang Guan 已提交
199
  if (mndAllocStep(pMnode, "mnode-stb", mndInitStb, mndCleanupStb) != 0) return -1;
S
sma  
Shengliang Guan 已提交
200
  if (mndAllocStep(pMnode, "mnode-stb", mndInitSma, mndCleanupSma) != 0) return -1;
D
dapan1121 已提交
201
  if (mndAllocStep(pMnode, "mnode-infos", mndInitInfos, mndCleanupInfos) != 0) return -1;
D
dapan1121 已提交
202
  if (mndAllocStep(pMnode, "mnode-perfs", mndInitPerfs, mndCleanupPerfs) != 0) return -1;
S
Shengliang Guan 已提交
203
  if (mndAllocStep(pMnode, "mnode-db", mndInitDb, mndCleanupDb) != 0) return -1;
S
Shengliang Guan 已提交
204
  if (mndAllocStep(pMnode, "mnode-func", mndInitFunc, mndCleanupFunc) != 0) return -1;
S
shm  
Shengliang Guan 已提交
205
  if (deploy) {
S
Shengliang Guan 已提交
206 207 208 209 210 211
    if (mndAllocStep(pMnode, "mnode-sdb-deploy", mndDeploySdb, NULL) != 0) return -1;
  } else {
    if (mndAllocStep(pMnode, "mnode-sdb-read", mndReadSdb, NULL) != 0) return -1;
  }
  if (mndAllocStep(pMnode, "mnode-profile", mndInitProfile, mndCleanupProfile) != 0) return -1;
  if (mndAllocStep(pMnode, "mnode-show", mndInitShow, mndCleanupShow) != 0) return -1;
D
dapan1121 已提交
212
  if (mndAllocStep(pMnode, "mnode-query", mndInitQuery, mndCleanupQuery) != 0) return -1;
S
Shengliang Guan 已提交
213 214
  if (mndAllocStep(pMnode, "mnode-sync", mndInitSync, mndCleanupSync) != 0) return -1;
  if (mndAllocStep(pMnode, "mnode-telem", mndInitTelem, mndCleanupTelem) != 0) return -1;
S
Shengliang Guan 已提交
215 216 217 218

  return 0;
}

219
static void mndCleanupSteps(SMnode *pMnode, int32_t pos) {
S
Shengliang Guan 已提交
220 221
  if (pMnode->pSteps == NULL) return;

222
  if (pos == -1) {
223
    pos = taosArrayGetSize(pMnode->pSteps) - 1;
S
Shengliang Guan 已提交
224 225
  }

226
  for (int32_t s = pos; s >= 0; s--) {
227
    SMnodeStep *pStep = taosArrayGet(pMnode->pSteps, s);
S
Shengliang Guan 已提交
228
    mDebug("%s will cleanup", pStep->name);
229 230 231
    if (pStep->cleanupFp != NULL) {
      (*pStep->cleanupFp)(pMnode);
    }
S
Shengliang Guan 已提交
232 233
  }

S
Shengliang Guan 已提交
234
  taosArrayClear(pMnode->pSteps);
S
Shengliang Guan 已提交
235
  taosArrayDestroy(pMnode->pSteps);
S
Shengliang Guan 已提交
236
  pMnode->pSteps = NULL;
237
}
S
Shengliang Guan 已提交
238

239
static int32_t mndExecSteps(SMnode *pMnode) {
S
Shengliang Guan 已提交
240
  int32_t size = taosArrayGetSize(pMnode->pSteps);
241
  for (int32_t pos = 0; pos < size; pos++) {
S
Shengliang Guan 已提交
242
    SMnodeStep *pStep = taosArrayGet(pMnode->pSteps, pos);
243
    if (pStep->initFp == NULL) continue;
S
Shengliang Guan 已提交
244

S
Shengliang Guan 已提交
245
    if ((*pStep->initFp)(pMnode) != 0) {
S
Shengliang Guan 已提交
246
      int32_t code = terrno;
S
Shengliang Guan 已提交
247
      mError("%s exec failed since %s, start to cleanup", pStep->name, terrstr());
248
      mndCleanupSteps(pMnode, pos);
S
Shengliang Guan 已提交
249
      terrno = code;
S
Shengliang Guan 已提交
250
      return -1;
S
Shengliang Guan 已提交
251
    } else {
S
Shengliang Guan 已提交
252
      mDebug("%s is initialized", pStep->name);
S
Shengliang Guan 已提交
253
      tmsgReportStartup(pStep->name, "initialized");
S
Shengliang Guan 已提交
254 255
    }
  }
S
Shengliang Guan 已提交
256

S
shm  
Shengliang Guan 已提交
257
  pMnode->clusterId = mndGetClusterId(pMnode);
S
Shengliang Guan 已提交
258
  return 0;
259
}
S
Shengliang Guan 已提交
260

S
shm  
Shengliang Guan 已提交
261
static void mndSetOptions(SMnode *pMnode, const SMnodeOpt *pOption) {
262 263 264
  pMnode->replica = pOption->replica;
  pMnode->selfIndex = pOption->selfIndex;
  memcpy(&pMnode->replicas, pOption->replicas, sizeof(SReplica) * TSDB_MAX_REPLICA);
S
Shengliang Guan 已提交
265
  pMnode->msgCb = pOption->msgCb;
266
  pMnode->selfId = pOption->replicas[pOption->selfIndex].id;
L
Liu Jicong 已提交
267
}
268

269
SMnode *mndOpen(const char *path, const SMnodeOpt *pOption) {
S
Shengliang Guan 已提交
270 271
  mDebug("start to open mnode in %s", path);

wafwerar's avatar
wafwerar 已提交
272
  SMnode *pMnode = taosMemoryCalloc(1, sizeof(SMnode));
S
Shengliang Guan 已提交
273 274 275 276 277 278
  if (pMnode == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    mError("failed to open mnode since %s", terrstr());
    return NULL;
  }

S
Shengliang Guan 已提交
279 280
  char timestr[24] = "1970-01-01 00:00:00.00";
  (void)taosParseTime(timestr, &pMnode->checkTime, (int32_t)strlen(timestr), TSDB_TIME_PRECISION_MILLI, 0);
S
shm  
Shengliang Guan 已提交
281
  mndSetOptions(pMnode, pOption);
S
Shengliang Guan 已提交
282

S
Shengliang Guan 已提交
283 284
  pMnode->pSteps = taosArrayInit(24, sizeof(SMnodeStep));
  if (pMnode->pSteps == NULL) {
wafwerar's avatar
wafwerar 已提交
285
    taosMemoryFree(pMnode);
S
Shengliang Guan 已提交
286 287 288 289
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    mError("failed to open mnode since %s", terrstr());
    return NULL;
  }
290

S
Shengliang Guan 已提交
291
  int32_t code = mndCreateDir(pMnode, path);
292
  if (code != 0) {
S
Shengliang Guan 已提交
293 294
    code = terrno;
    mError("failed to open mnode since %s", terrstr());
295 296 297 298 299
    mndClose(pMnode);
    terrno = code;
    return NULL;
  }

S
shm  
Shengliang Guan 已提交
300
  code = mndInitSteps(pMnode, pOption->deploy);
301
  if (code != 0) {
S
Shengliang Guan 已提交
302 303
    code = terrno;
    mError("failed to open mnode since %s", terrstr());
304 305 306 307 308 309 310
    mndClose(pMnode);
    terrno = code;
    return NULL;
  }

  code = mndExecSteps(pMnode);
  if (code != 0) {
S
Shengliang Guan 已提交
311 312
    code = terrno;
    mError("failed to open mnode since %s", terrstr());
313 314 315 316
    mndClose(pMnode);
    terrno = code;
    return NULL;
  }
S
Shengliang Guan 已提交
317

S
Shengliang Guan 已提交
318
  mndUpdateMnodeRole(pMnode);
S
Shengliang Guan 已提交
319
  mDebug("mnode open successfully ");
S
Shengliang Guan 已提交
320 321
  return pMnode;
}
S
Shengliang Guan 已提交
322

323
void mndClose(SMnode *pMnode) {
S
Shengliang Guan 已提交
324 325 326
  if (pMnode != NULL) {
    mDebug("start to close mnode");
    mndCleanupSteps(pMnode, -1);
wafwerar's avatar
wafwerar 已提交
327 328
    taosMemoryFreeClear(pMnode->path);
    taosMemoryFreeClear(pMnode);
S
Shengliang Guan 已提交
329 330
    mDebug("mnode is closed");
  }
331
}
S
Shengliang Guan 已提交
332

333
int32_t mndAlter(SMnode *pMnode, const SMnodeOpt *pOption) {
S
Shengliang Guan 已提交
334 335
  mDebug("start to alter mnode");
  mDebug("mnode is altered");
336 337
  return 0;
}
S
Shengliang Guan 已提交
338

M
Minghao Li 已提交
339 340 341 342 343 344 345 346 347 348 349
int32_t mndStart(SMnode *pMnode) { 
  syncStart(pMnode->syncMgmt.sync); 
  return mndInitTimer(pMnode);
}

void mndStop(SMnode *pMnode) { 
  syncStop(pMnode->syncMgmt.sync); 
  return mndCleanupTimer(pMnode); 
}

int32_t mndProcessApplyMsg(SRpcMsg *pMsg) {
S
Shengliang Guan 已提交
350

M
Minghao Li 已提交
351 352 353 354 355 356 357
  SSdbRaw *pRaw = pMsg->pCont;
  SMnode *pMnode = pMsg->info.node;
  int32_t code = sdbWriteWithoutFree(pMnode->pSdb, pRaw);
  rpcFreeCont(pMsg->pCont);

  return code;
}
S
Shengliang Guan 已提交
358

M
Minghao Li 已提交
359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 431 432 433 434 435 436 437 438 439 440 441 442 443 444 445 446 447 448 449 450 451 452 453 454 455 456
#include "syncTools.h"

int32_t mndProcessSyncMsg(SRpcMsg *pMsg) {
  SMnode *pMnode = pMsg->info.node;
  void   *ahandle = pMsg->info.ahandle;

  int32_t ret = TAOS_SYNC_PROPOSE_OTHER_ERROR;
  
  if (syncEnvIsStart()) {
    SSyncNode *pSyncNode = syncNodeAcquire(pMnode->syncMgmt.sync);
    assert(pSyncNode != NULL);

    ESyncState state = syncGetMyRole(pMnode->syncMgmt.sync);
    SyncTerm   currentTerm = syncGetMyTerm(pMnode->syncMgmt.sync);

    SMsgHead *pHead = pMsg->pCont;

    char  logBuf[512];
    char *syncNodeStr = sync2SimpleStr(pMnode->syncMgmt.sync);
    snprintf(logBuf, sizeof(logBuf), "==vnodeProcessSyncReq== msgType:%d, syncNode: %s", pMsg->msgType, syncNodeStr);
    syncRpcMsgLog2(logBuf, pMsg);
    taosMemoryFree(syncNodeStr);

    SRpcMsg *pRpcMsg = pMsg;

    if (pRpcMsg->msgType == TDMT_VND_SYNC_TIMEOUT) {
      SyncTimeout *pSyncMsg = syncTimeoutFromRpcMsg2(pRpcMsg);
      assert(pSyncMsg != NULL);

      ret = syncNodeOnTimeoutCb(pSyncNode, pSyncMsg);
      syncTimeoutDestroy(pSyncMsg);

    } else if (pRpcMsg->msgType == TDMT_VND_SYNC_PING) {
      SyncPing *pSyncMsg = syncPingFromRpcMsg2(pRpcMsg);
      assert(pSyncMsg != NULL);

      ret = syncNodeOnPingCb(pSyncNode, pSyncMsg);
      syncPingDestroy(pSyncMsg);

    } else if (pRpcMsg->msgType == TDMT_VND_SYNC_PING_REPLY) {
      SyncPingReply *pSyncMsg = syncPingReplyFromRpcMsg2(pRpcMsg);
      assert(pSyncMsg != NULL);

      ret = syncNodeOnPingReplyCb(pSyncNode, pSyncMsg);
      syncPingReplyDestroy(pSyncMsg);

    } else if (pRpcMsg->msgType == TDMT_VND_SYNC_CLIENT_REQUEST) {
      SyncClientRequest *pSyncMsg = syncClientRequestFromRpcMsg2(pRpcMsg);
      assert(pSyncMsg != NULL);

      ret = syncNodeOnClientRequestCb(pSyncNode, pSyncMsg);
      syncClientRequestDestroy(pSyncMsg);

    } else if (pRpcMsg->msgType == TDMT_VND_SYNC_REQUEST_VOTE) {
      SyncRequestVote *pSyncMsg = syncRequestVoteFromRpcMsg2(pRpcMsg);
      assert(pSyncMsg != NULL);

      ret = syncNodeOnRequestVoteCb(pSyncNode, pSyncMsg);
      syncRequestVoteDestroy(pSyncMsg);

    } else if (pRpcMsg->msgType == TDMT_VND_SYNC_REQUEST_VOTE_REPLY) {
      SyncRequestVoteReply *pSyncMsg = syncRequestVoteReplyFromRpcMsg2(pRpcMsg);
      assert(pSyncMsg != NULL);

      ret = syncNodeOnRequestVoteReplyCb(pSyncNode, pSyncMsg);
      syncRequestVoteReplyDestroy(pSyncMsg);

    } else if (pRpcMsg->msgType == TDMT_VND_SYNC_APPEND_ENTRIES) {
      SyncAppendEntries *pSyncMsg = syncAppendEntriesFromRpcMsg2(pRpcMsg);
      assert(pSyncMsg != NULL);

      ret = syncNodeOnAppendEntriesCb(pSyncNode, pSyncMsg);
      syncAppendEntriesDestroy(pSyncMsg);

    } else if (pRpcMsg->msgType == TDMT_VND_SYNC_APPEND_ENTRIES_REPLY) {
      SyncAppendEntriesReply *pSyncMsg = syncAppendEntriesReplyFromRpcMsg2(pRpcMsg);
      assert(pSyncMsg != NULL);

      ret = syncNodeOnAppendEntriesReplyCb(pSyncNode, pSyncMsg);
      syncAppendEntriesReplyDestroy(pSyncMsg);

    } else {
      mError("==mndProcessSyncMsg== error msg type:%d", pRpcMsg->msgType);
      ret = TAOS_SYNC_PROPOSE_OTHER_ERROR;
    }

    syncNodeRelease(pSyncNode);
  } else {
    mError("==mndProcessSyncMsg== error syncEnv stop");
    ret = TAOS_SYNC_PROPOSE_OTHER_ERROR;
  }

  return ret;


  return 0;
}

S
Shengliang Guan 已提交
457 458 459 460
int32_t mndProcessMsg(SRpcMsg *pMsg) {
  SMnode *pMnode = pMsg->info.node;
  void   *ahandle = pMsg->info.ahandle;
  mTrace("msg:%p, will be processed, type:%s app:%p", pMsg, TMSG_INFO(pMsg->msgType), ahandle);
S
Shengliang Guan 已提交
461

S
Shengliang Guan 已提交
462 463 464 465 466 467
  if (IsReq(pMsg)) {
    if (!mndIsMaster(pMnode)) {
      terrno = TSDB_CODE_APP_NOT_READY;
      mDebug("msg:%p, failed to process since %s, app:%p", pMsg, terrstr(), ahandle);
      return -1;
    }
S
Shengliang Guan 已提交
468

S
Shengliang Guan 已提交
469 470 471 472 473
    if (pMsg->contLen == 0 || pMsg->pCont == NULL) {
      terrno = TSDB_CODE_INVALID_MSG_LEN;
      mError("msg:%p, failed to process since %s, app:%p", pMsg, terrstr(), ahandle);
      return -1;
    }
S
Shengliang Guan 已提交
474 475
  }

S
Shengliang Guan 已提交
476
  MndMsgFp fp = pMnode->msgFp[TMSG_INDEX(pMsg->msgType)];
S
Shengliang Guan 已提交
477
  if (fp == NULL) {
S
shm  
Shengliang Guan 已提交
478
    terrno = TSDB_CODE_MSG_NOT_PROCESSED;
S
Shengliang Guan 已提交
479
    mError("msg:%p, failed to process since no msg handle, app:%p", pMsg, ahandle);
S
shm  
Shengliang Guan 已提交
480
    return -1;
S
Shengliang Guan 已提交
481 482
  }

S
shm  
Shengliang Guan 已提交
483
  int32_t code = (*fp)(pMsg);
S
Shengliang Guan 已提交
484
  if (code == TSDB_CODE_ACTION_IN_PROGRESS) {
S
shm  
Shengliang Guan 已提交
485
    terrno = code;
S
Shengliang Guan 已提交
486
    mTrace("msg:%p, in progress, app:%p", pMsg, ahandle);
S
Shengliang Guan 已提交
487
  } else if (code != 0) {
488 489 490 491 492
    if (terrno != TSDB_CODE_OPS_NOT_SUPPORT) {
      mError("msg:%p, failed to process since %s, app:%p", pMsg, terrstr(), ahandle);
    } else {
      mTrace("msg:%p, failed to process since %s, app:%p", pMsg, terrstr(), ahandle);
    }
493
  } else {
S
Shengliang Guan 已提交
494
    mTrace("msg:%p, is processed, app:%p", pMsg, ahandle);
S
Shengliang Guan 已提交
495 496
  }

S
shm  
Shengliang Guan 已提交
497
  return code;
S
Shengliang Guan 已提交
498 499
}

500 501 502 503
void mndSetMsgHandle(SMnode *pMnode, tmsg_t msgType, MndMsgFp fp) {
  tmsg_t type = TMSG_INDEX(msgType);
  if (type >= 0 && type < TDMT_MAX) {
    pMnode->msgFp[type] = fp;
S
Shengliang Guan 已提交
504 505 506
  }
}

D
dapan1121 已提交
507
// Note: uid 0 is reserved
S
sma  
Shengliang Guan 已提交
508
int64_t mndGenerateUid(char *name, int32_t len) {
S
monitor  
Shengliang Guan 已提交
509
  int32_t hashval = MurmurHash3_32(name, len);
D
dapan1121 已提交
510 511

  do {
L
Liu Jicong 已提交
512
    int64_t us = taosGetTimestampUs();
S
sma  
Shengliang Guan 已提交
513 514
    int64_t x = (us & 0x000000FFFFFFFFFF) << 24;
    int64_t uuid = x + ((hashval & ((1ul << 16) - 1ul)) << 8) + (taosRand() & ((1ul << 8) - 1ul));
D
dapan1121 已提交
515
    if (uuid) {
L
Liu Jicong 已提交
516
      return llabs(uuid);
D
dapan1121 已提交
517 518
    }
  } while (true);
L
Liu Jicong 已提交
519
}
S
Shengliang Guan 已提交
520 521 522

int32_t mndGetMonitorInfo(SMnode *pMnode, SMonClusterInfo *pClusterInfo, SMonVgroupInfo *pVgroupInfo,
                          SMonGrantInfo *pGrantInfo) {
S
Shengliang Guan 已提交
523
  if (!mndIsMaster(pMnode)) return -1;
S
Shengliang Guan 已提交
524

S
Shengliang Guan 已提交
525 526 527 528 529 530 531 532 533 534 535 536 537 538 539 540 541 542 543 544 545 546 547 548 549 550 551 552 553 554 555 556 557 558 559 560 561 562 563 564 565 566
  SSdb   *pSdb = pMnode->pSdb;
  int64_t ms = taosGetTimestampMs();

  pClusterInfo->dnodes = taosArrayInit(sdbGetSize(pSdb, SDB_DNODE), sizeof(SMonDnodeDesc));
  pClusterInfo->mnodes = taosArrayInit(sdbGetSize(pSdb, SDB_MNODE), sizeof(SMonMnodeDesc));
  pVgroupInfo->vgroups = taosArrayInit(sdbGetSize(pSdb, SDB_VGROUP), sizeof(SMonVgroupDesc));
  if (pClusterInfo->dnodes == NULL || pClusterInfo->mnodes == NULL || pVgroupInfo->vgroups == NULL) {
    return -1;
  }

  // cluster info
  tstrncpy(pClusterInfo->version, version, sizeof(pClusterInfo->version));
  pClusterInfo->monitor_interval = tsMonitorInterval;
  pClusterInfo->connections_total = mndGetNumOfConnections(pMnode);

  void *pIter = NULL;
  while (1) {
    SDnodeObj *pObj = NULL;
    pIter = sdbFetch(pSdb, SDB_DNODE, pIter, (void **)&pObj);
    if (pIter == NULL) break;

    SMonDnodeDesc desc = {0};
    desc.dnode_id = pObj->id;
    tstrncpy(desc.dnode_ep, pObj->ep, sizeof(desc.dnode_ep));
    if (mndIsDnodeOnline(pMnode, pObj, ms)) {
      tstrncpy(desc.status, "ready", sizeof(desc.status));
    } else {
      tstrncpy(desc.status, "offline", sizeof(desc.status));
    }
    taosArrayPush(pClusterInfo->dnodes, &desc);
    sdbRelease(pSdb, pObj);
  }

  pIter = NULL;
  while (1) {
    SMnodeObj *pObj = NULL;
    pIter = sdbFetch(pSdb, SDB_MNODE, pIter, (void **)&pObj);
    if (pIter == NULL) break;

    SMonMnodeDesc desc = {0};
    desc.mnode_id = pObj->id;
    tstrncpy(desc.mnode_ep, pObj->pDnode->ep, sizeof(desc.mnode_ep));
S
Shengliang Guan 已提交
567
    tstrncpy(desc.role, syncStr(pObj->role), sizeof(desc.role));
S
Shengliang Guan 已提交
568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588
    taosArrayPush(pClusterInfo->mnodes, &desc);
    sdbRelease(pSdb, pObj);

    if (pObj->role == TAOS_SYNC_STATE_LEADER) {
      pClusterInfo->first_ep_dnode_id = pObj->id;
      tstrncpy(pClusterInfo->first_ep, pObj->pDnode->ep, sizeof(pClusterInfo->first_ep));
      pClusterInfo->master_uptime = (ms - pObj->roleTime) / (86400000.0f);
    }
  }

  // vgroup info
  pIter = NULL;
  while (1) {
    SVgObj *pVgroup = NULL;
    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
    if (pIter == NULL) break;

    pClusterInfo->vgroups_total++;

    SMonVgroupDesc desc = {0};
    desc.vgroup_id = pVgroup->vgId;
589 590 591 592 593

    SName name = {0};
    tNameFromString(&name, pVgroup->dbName, T_NAME_ACCT | T_NAME_DB | T_NAME_TABLE);
    tNameGetDbName(&name, desc.database_name);

S
Shengliang Guan 已提交
594 595 596 597 598 599 600
    desc.tables_num = pVgroup->numOfTables;
    pGrantInfo->timeseries_used += pVgroup->numOfTimeSeries;
    tstrncpy(desc.status, "unsynced", sizeof(desc.status));
    for (int32_t i = 0; i < pVgroup->replica; ++i) {
      SVnodeGid     *pVgid = &pVgroup->vnodeGid[i];
      SMonVnodeDesc *pVnDesc = &desc.vnodes[i];
      pVnDesc->dnode_id = pVgid->dnodeId;
S
Shengliang Guan 已提交
601
      tstrncpy(pVnDesc->vnode_role, syncStr(pVgid->role), sizeof(pVnDesc->vnode_role));
S
Shengliang Guan 已提交
602 603 604 605
      if (pVgid->role == TAOS_SYNC_STATE_LEADER) {
        tstrncpy(desc.status, "ready", sizeof(desc.status));
        pClusterInfo->vgroups_alive++;
      }
S
Shengliang Guan 已提交
606
      if (pVgid->role != TAOS_SYNC_STATE_ERROR) {
S
Shengliang Guan 已提交
607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624
        pClusterInfo->vnodes_alive++;
      }
      pClusterInfo->vnodes_total++;
    }

    taosArrayPush(pVgroupInfo->vgroups, &desc);
    sdbRelease(pSdb, pVgroup);
  }

  // grant info
  pGrantInfo->expire_time = (pMnode->grant.expireTimeMS - ms) / 86400000.0f;
  pGrantInfo->timeseries_total = pMnode->grant.timeseriesAllowed;
  if (pMnode->grant.expireTimeMS == 0) {
    pGrantInfo->expire_time = INT32_MAX;
    pGrantInfo->timeseries_total = INT32_MAX;
  }

  return 0;
L
Liu Jicong 已提交
625
}
S
Shengliang Guan 已提交
626 627 628 629

int32_t mndGetLoad(SMnode *pMnode, SMnodeLoad *pLoad) {
  pLoad->syncState = pMnode->syncMgmt.state;
  return 0;
L
fix  
Liu Jicong 已提交
630
}