mnode.c 20.2 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"
S
monitor  
Shengliang Guan 已提交
25
#include "mndInfoSchema.h"
S
Shengliang Guan 已提交
26
#include "mndMnode.h"
L
Liu Jicong 已提交
27
#include "mndOffset.h"
S
Shengliang Guan 已提交
28
#include "mndProfile.h"
S
Shengliang Guan 已提交
29
#include "mndQnode.h"
S
Shengliang Guan 已提交
30
#include "mndShow.h"
S
Shengliang Guan 已提交
31
#include "mndSnode.h"
S
Shengliang Guan 已提交
32
#include "mndStb.h"
L
Liu Jicong 已提交
33
#include "mndSubscribe.h"
S
Shengliang Guan 已提交
34 35
#include "mndSync.h"
#include "mndTelem.h"
L
Liu Jicong 已提交
36
#include "mndTopic.h"
S
Shengliang Guan 已提交
37 38 39
#include "mndTrans.h"
#include "mndUser.h"
#include "mndVgroup.h"
S
Shengliang Guan 已提交
40

S
Shengliang Guan 已提交
41 42 43 44
#define MQ_TIMER_MS    3000
#define TRNAS_TIMER_MS 6000
#define TELEM_TIMER_MS 86400000

S
Shengliang Guan 已提交
45 46 47 48
int32_t mndSendReqToDnode(SMnode *pMnode, SEpSet *pEpSet, SRpcMsg *pMsg) {
  if (pMnode == NULL || pMnode->sendReqToDnodeFp == NULL) {
    terrno = TSDB_CODE_MND_NOT_READY;
    return -1;
49
  }
S
Shengliang Guan 已提交
50 51

  return (*pMnode->sendReqToDnodeFp)(pMnode->pDnode, pEpSet, pMsg);
S
Shengliang Guan 已提交
52
}
S
Shengliang Guan 已提交
53

S
Shengliang Guan 已提交
54 55 56 57
int32_t mndSendReqToMnode(SMnode *pMnode, SRpcMsg *pMsg) {
  if (pMnode == NULL || pMnode->sendReqToDnodeFp == NULL) {
    terrno = TSDB_CODE_MND_NOT_READY;
    return -1;
58
  }
S
Shengliang Guan 已提交
59 60

  return (*pMnode->sendReqToMnodeFp)(pMnode->pDnode, pMsg);
S
Shengliang Guan 已提交
61
}
S
Shengliang Guan 已提交
62

S
Shengliang Guan 已提交
63 64 65
void mndSendRedirectRsp(SMnode *pMnode, SRpcMsg *pMsg) {
  if (pMnode != NULL && pMnode->sendRedirectRspFp != NULL) {
    (*pMnode->sendRedirectRspFp)(pMnode->pDnode, pMsg);
66
  }
S
Shengliang Guan 已提交
67
}
S
Shengliang Guan 已提交
68

S
Shengliang Guan 已提交
69 70 71 72 73 74 75 76 77 78 79 80 81
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 已提交
82
static void mndPullupTrans(void *param, void *tmrId) {
S
Shengliang Guan 已提交
83 84
  SMnode *pMnode = param;
  if (mndIsMaster(pMnode)) {
S
Shengliang Guan 已提交
85 86
    int32_t contLen = 0;
    void   *pReq = mndBuildTimerMsg(&contLen);
S
Shengliang Guan 已提交
87
    SRpcMsg rpcMsg = {.msgType = TDMT_MND_TRANS_TIMER, .pCont = pReq, .contLen = contLen};
S
Shengliang Guan 已提交
88
    pMnode->putReqToMWriteQFp(pMnode->pDnode, &rpcMsg);
S
Shengliang Guan 已提交
89 90
  }

S
Shengliang Guan 已提交
91
  taosTmrReset(mndPullupTrans, TRNAS_TIMER_MS, pMnode, pMnode->timer, &pMnode->transTimer);
S
Shengliang Guan 已提交
92 93
}

L
Liu Jicong 已提交
94 95
static void mndCalMqRebalance(void *param, void *tmrId) {
  SMnode *pMnode = param;
L
Liu Jicong 已提交
96
  if (mndIsMaster(pMnode)) {
S
Shengliang Guan 已提交
97 98
    int32_t contLen = 0;
    void   *pReq = mndBuildTimerMsg(&contLen);
L
Liu Jicong 已提交
99 100 101 102 103
    SRpcMsg rpcMsg = {
        .msgType = TDMT_MND_MQ_TIMER,
        .pCont = pReq,
        .contLen = contLen,
    };
L
Liu Jicong 已提交
104
    pMnode->putReqToMReadQFp(pMnode->pDnode, &rpcMsg);
L
Liu Jicong 已提交
105 106
  }

S
Shengliang Guan 已提交
107 108 109
  taosTmrReset(mndCalMqRebalance, MQ_TIMER_MS, pMnode, pMnode->timer, &pMnode->mqTimer);
}

S
Shengliang Guan 已提交
110
static void mndPullupTelem(void *param, void *tmrId) {
S
Shengliang Guan 已提交
111 112 113 114 115 116 117 118
  SMnode *pMnode = param;
  if (mndIsMaster(pMnode)) {
    int32_t contLen = 0;
    void   *pReq = mndBuildTimerMsg(&contLen);
    SRpcMsg rpcMsg = {.msgType = TDMT_MND_TELEM_TIMER, .pCont = pReq, .contLen = contLen};
    pMnode->putReqToMReadQFp(pMnode->pDnode, &rpcMsg);
  }

S
Shengliang Guan 已提交
119
  taosTmrReset(mndPullupTelem, TELEM_TIMER_MS, pMnode, pMnode->timer, &pMnode->telemTimer);
L
Liu Jicong 已提交
120 121
}

122
static int32_t mndInitTimer(SMnode *pMnode) {
S
Shengliang Guan 已提交
123
  pMnode->timer = taosTmrInit(5000, 200, 3600000, "MND");
124
  if (pMnode->timer == NULL) {
S
Shengliang Guan 已提交
125 126
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return -1;
S
Shengliang Guan 已提交
127 128
  }

S
Shengliang Guan 已提交
129
  if (taosTmrReset(mndPullupTrans, TRNAS_TIMER_MS, pMnode, pMnode->timer, &pMnode->transTimer)) {
S
Shengliang Guan 已提交
130
    terrno = TSDB_CODE_OUT_OF_MEMORY;
S
Shengliang Guan 已提交
131 132 133
    return -1;
  }

S
Shengliang Guan 已提交
134
  if (taosTmrReset(mndCalMqRebalance, MQ_TIMER_MS, pMnode, pMnode->timer, &pMnode->mqTimer)) {
S
Shengliang Guan 已提交
135 136 137 138
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return -1;
  }

S
Shengliang Guan 已提交
139
  if (taosTmrReset(mndPullupTelem, 60000, pMnode, pMnode->timer, &pMnode->telemTimer)) {
L
Liu Jicong 已提交
140 141 142 143
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return -1;
  }

S
Shengliang Guan 已提交
144 145 146
  return 0;
}

147 148
static void mndCleanupTimer(SMnode *pMnode) {
  if (pMnode->timer != NULL) {
S
Shengliang Guan 已提交
149 150
    taosTmrStop(pMnode->transTimer);
    pMnode->transTimer = NULL;
L
Liu Jicong 已提交
151 152
    taosTmrStop(pMnode->mqTimer);
    pMnode->mqTimer = NULL;
S
Shengliang Guan 已提交
153 154
    taosTmrStop(pMnode->telemTimer);
    pMnode->telemTimer = NULL;
155 156
    taosTmrCleanUp(pMnode->timer);
    pMnode->timer = NULL;
S
Shengliang Guan 已提交
157 158 159
  }
}

S
Shengliang Guan 已提交
160
static int32_t mndCreateDir(SMnode *pMnode, const char *path) {
161 162 163
  pMnode->path = strdup(path);
  if (pMnode->path == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
S
Shengliang Guan 已提交
164
    return -1;
165 166 167 168
  }

  if (taosMkDir(pMnode->path) != 0) {
    terrno = TAOS_SYSTEM_ERROR(errno);
S
Shengliang Guan 已提交
169
    return -1;
170
  }
171 172

  return 0;
173
}
S
Shengliang Guan 已提交
174

175 176 177
static int32_t mndInitSdb(SMnode *pMnode) {
  SSdbOpt opt = {0};
  opt.path = pMnode->path;
S
Shengliang Guan 已提交
178
  opt.pMnode = pMnode;
S
Shengliang Guan 已提交
179

S
Shengliang Guan 已提交
180
  pMnode->pSdb = sdbInit(&opt);
181
  if (pMnode->pSdb == NULL) {
S
Shengliang Guan 已提交
182 183 184 185 186 187
    return -1;
  }

  return 0;
}

188
static int32_t mndDeploySdb(SMnode *pMnode) { return sdbDeploy(pMnode->pSdb); }
S
Shengliang Guan 已提交
189
static int32_t mndReadSdb(SMnode *pMnode) { return sdbReadFile(pMnode->pSdb); }
190 191 192

static void mndCleanupSdb(SMnode *pMnode) {
  if (pMnode->pSdb) {
S
Shengliang Guan 已提交
193
    sdbCleanup(pMnode->pSdb);
194 195 196 197
    pMnode->pSdb = NULL;
  }
}

198 199 200 201 202
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 已提交
203
  if (taosArrayPush(pMnode->pSteps, &step) == NULL) {
204 205 206 207
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return -1;
  }

S
Shengliang Guan 已提交
208 209 210
  return 0;
}

211
static int32_t mndInitSteps(SMnode *pMnode) {
S
Shengliang Guan 已提交
212 213 214 215
  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 已提交
216 217 218
  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 已提交
219
  if (mndAllocStep(pMnode, "mnode-dnode", mndInitDnode, mndCleanupDnode) != 0) return -1;
S
Shengliang Guan 已提交
220
  if (mndAllocStep(pMnode, "mnode-user", mndInitUser, mndCleanupUser) != 0) return -1;
S
Shengliang Guan 已提交
221 222 223
  if (mndAllocStep(pMnode, "mnode-auth", mndInitAuth, mndCleanupAuth) != 0) return -1;
  if (mndAllocStep(pMnode, "mnode-acct", mndInitAcct, mndCleanupAcct) != 0) return -1;
  if (mndAllocStep(pMnode, "mnode-topic", mndInitTopic, mndCleanupTopic) != 0) return -1;
L
Liu Jicong 已提交
224 225
  if (mndAllocStep(pMnode, "mnode-consumer", mndInitConsumer, mndCleanupConsumer) != 0) return -1;
  if (mndAllocStep(pMnode, "mnode-subscribe", mndInitSubscribe, mndCleanupSubscribe) != 0) return -1;
L
Liu Jicong 已提交
226
  if (mndAllocStep(pMnode, "mnode-offset", mndInitOffset, mndCleanupOffset) != 0) return -1;
S
Shengliang Guan 已提交
227
  if (mndAllocStep(pMnode, "mnode-vgroup", mndInitVgroup, mndCleanupVgroup) != 0) return -1;
S
Shengliang Guan 已提交
228
  if (mndAllocStep(pMnode, "mnode-stb", mndInitStb, mndCleanupStb) != 0) return -1;
D
dapan1121 已提交
229
  if (mndAllocStep(pMnode, "mnode-infos", mndInitInfos, mndCleanupInfos) != 0) return -1;
S
Shengliang Guan 已提交
230
  if (mndAllocStep(pMnode, "mnode-db", mndInitDb, mndCleanupDb) != 0) return -1;
S
Shengliang Guan 已提交
231
  if (mndAllocStep(pMnode, "mnode-func", mndInitFunc, mndCleanupFunc) != 0) return -1;
232
  if (pMnode->clusterId <= 0) {
S
Shengliang Guan 已提交
233 234 235 236 237 238 239 240 241 242
    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-timer", mndInitTimer, NULL) != 0) return -1;
  if (mndAllocStep(pMnode, "mnode-profile", mndInitProfile, mndCleanupProfile) != 0) return -1;
  if (mndAllocStep(pMnode, "mnode-show", mndInitShow, mndCleanupShow) != 0) return -1;
  if (mndAllocStep(pMnode, "mnode-sync", mndInitSync, mndCleanupSync) != 0) return -1;
  if (mndAllocStep(pMnode, "mnode-telem", mndInitTelem, mndCleanupTelem) != 0) return -1;
  if (mndAllocStep(pMnode, "mnode-timer", NULL, mndCleanupTimer) != 0) return -1;
S
Shengliang Guan 已提交
243 244 245 246

  return 0;
}

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

250
  if (pos == -1) {
251
    pos = taosArrayGetSize(pMnode->pSteps) - 1;
S
Shengliang Guan 已提交
252 253
  }

254
  for (int32_t s = pos; s >= 0; s--) {
255
    SMnodeStep *pStep = taosArrayGet(pMnode->pSteps, s);
S
Shengliang Guan 已提交
256
    mDebug("%s will cleanup", pStep->name);
257 258 259
    if (pStep->cleanupFp != NULL) {
      (*pStep->cleanupFp)(pMnode);
    }
S
Shengliang Guan 已提交
260 261
  }

S
Shengliang Guan 已提交
262
  taosArrayClear(pMnode->pSteps);
S
Shengliang Guan 已提交
263
  taosArrayDestroy(pMnode->pSteps);
S
Shengliang Guan 已提交
264
  pMnode->pSteps = NULL;
265
}
S
Shengliang Guan 已提交
266

267
static int32_t mndExecSteps(SMnode *pMnode) {
S
Shengliang Guan 已提交
268
  int32_t size = taosArrayGetSize(pMnode->pSteps);
269
  for (int32_t pos = 0; pos < size; pos++) {
S
Shengliang Guan 已提交
270
    SMnodeStep *pStep = taosArrayGet(pMnode->pSteps, pos);
271
    if (pStep->initFp == NULL) continue;
S
Shengliang Guan 已提交
272

S
Shengliang Guan 已提交
273
    if ((*pStep->initFp)(pMnode) != 0) {
S
Shengliang Guan 已提交
274
      int32_t code = terrno;
S
Shengliang Guan 已提交
275
      mError("%s exec failed since %s, start to cleanup", pStep->name, terrstr());
276
      mndCleanupSteps(pMnode, pos);
S
Shengliang Guan 已提交
277
      terrno = code;
S
Shengliang Guan 已提交
278
      return -1;
S
Shengliang Guan 已提交
279
    } else {
S
Shengliang Guan 已提交
280
      mDebug("%s is initialized", pStep->name);
S
Shengliang Guan 已提交
281 282
    }
  }
S
Shengliang Guan 已提交
283 284

  return 0;
285
}
S
Shengliang Guan 已提交
286

287 288 289 290 291 292 293
static int32_t mndSetOptions(SMnode *pMnode, const SMnodeOpt *pOption) {
  pMnode->dnodeId = pOption->dnodeId;
  pMnode->clusterId = pOption->clusterId;
  pMnode->replica = pOption->replica;
  pMnode->selfIndex = pOption->selfIndex;
  memcpy(&pMnode->replicas, pOption->replicas, sizeof(SReplica) * TSDB_MAX_REPLICA);
  pMnode->pDnode = pOption->pDnode;
S
Shengliang Guan 已提交
294
  pMnode->putReqToMWriteQFp = pOption->putReqToMWriteQFp;
L
Liu Jicong 已提交
295
  pMnode->putReqToMReadQFp = pOption->putReqToMReadQFp;
S
Shengliang Guan 已提交
296 297 298
  pMnode->sendReqToDnodeFp = pOption->sendReqToDnodeFp;
  pMnode->sendReqToMnodeFp = pOption->sendReqToMnodeFp;
  pMnode->sendRedirectRspFp = pOption->sendRedirectRspFp;
299

S
Shengliang Guan 已提交
300
  if (pMnode->sendReqToDnodeFp == NULL || pMnode->sendReqToMnodeFp == NULL || pMnode->sendRedirectRspFp == NULL ||
S
Shengliang Guan 已提交
301
      pMnode->putReqToMWriteQFp == NULL || pMnode->dnodeId < 0 || pMnode->clusterId < 0) {
S
Shengliang Guan 已提交
302
    terrno = TSDB_CODE_MND_INVALID_OPTIONS;
S
Shengliang Guan 已提交
303 304 305
    return -1;
  }

306
  return 0;
L
Liu Jicong 已提交
307
}
308

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

312
  SMnode *pMnode = calloc(1, sizeof(SMnode));
S
Shengliang Guan 已提交
313 314 315 316 317 318
  if (pMnode == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    mError("failed to open mnode since %s", terrstr());
    return NULL;
  }

S
Shengliang Guan 已提交
319 320 321
  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
Shengliang Guan 已提交
322 323 324 325 326 327 328
  pMnode->pSteps = taosArrayInit(24, sizeof(SMnodeStep));
  if (pMnode->pSteps == NULL) {
    free(pMnode);
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    mError("failed to open mnode since %s", terrstr());
    return NULL;
  }
329

S
Shengliang Guan 已提交
330
  int32_t code = mndCreateDir(pMnode, path);
331
  if (code != 0) {
S
Shengliang Guan 已提交
332 333
    code = terrno;
    mError("failed to open mnode since %s", terrstr());
334 335 336 337 338 339
    mndClose(pMnode);
    terrno = code;
    return NULL;
  }

  code = mndSetOptions(pMnode, pOption);
340
  if (code != 0) {
S
Shengliang Guan 已提交
341 342
    code = terrno;
    mError("failed to open mnode since %s", terrstr());
343 344 345 346 347 348 349
    mndClose(pMnode);
    terrno = code;
    return NULL;
  }

  code = mndInitSteps(pMnode);
  if (code != 0) {
S
Shengliang Guan 已提交
350 351
    code = terrno;
    mError("failed to open mnode since %s", terrstr());
352 353 354 355 356 357 358
    mndClose(pMnode);
    terrno = code;
    return NULL;
  }

  code = mndExecSteps(pMnode);
  if (code != 0) {
S
Shengliang Guan 已提交
359 360
    code = terrno;
    mError("failed to open mnode since %s", terrstr());
361 362 363 364
    mndClose(pMnode);
    terrno = code;
    return NULL;
  }
S
Shengliang Guan 已提交
365

S
Shengliang Guan 已提交
366
  mndUpdateMnodeRole(pMnode);
S
Shengliang Guan 已提交
367
  mDebug("mnode open successfully ");
S
Shengliang Guan 已提交
368 369
  return pMnode;
}
S
Shengliang Guan 已提交
370

371
void mndClose(SMnode *pMnode) {
S
Shengliang Guan 已提交
372 373 374 375 376 377 378
  if (pMnode != NULL) {
    mDebug("start to close mnode");
    mndCleanupSteps(pMnode, -1);
    tfree(pMnode->path);
    tfree(pMnode);
    mDebug("mnode is closed");
  }
379
}
S
Shengliang Guan 已提交
380

381
int32_t mndAlter(SMnode *pMnode, const SMnodeOpt *pOption) {
S
Shengliang Guan 已提交
382 383
  mDebug("start to alter mnode");
  mDebug("mnode is altered");
384 385
  return 0;
}
S
Shengliang Guan 已提交
386

387
void mndDestroy(const char *path) {
S
Shengliang Guan 已提交
388
  mDebug("start to destroy mnode at %s", path);
S
Shengliang Guan 已提交
389
  taosRemoveDir(path);
S
Shengliang Guan 已提交
390
  mDebug("mnode is destroyed");
391
}
S
Shengliang Guan 已提交
392

393
SMnodeMsg *mndInitMsg(SMnode *pMnode, SRpcMsg *pRpcMsg) {
S
Shengliang Guan 已提交
394 395 396
  SMnodeMsg *pMsg = taosAllocateQitem(sizeof(SMnodeMsg));
  if (pMsg == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
S
Shengliang Guan 已提交
397
    mError("failed to create msg since %s, app:%p RPC:%p", terrstr(), pRpcMsg->ahandle, pRpcMsg->handle);
S
Shengliang Guan 已提交
398 399 400
    return NULL;
  }

S
Shengliang Guan 已提交
401 402
  if (pRpcMsg->msgType != TDMT_MND_TRANS_TIMER && pRpcMsg->msgType != TDMT_MND_MQ_TIMER &&
      pRpcMsg->msgType != TDMT_MND_MQ_DO_REBALANCE && pRpcMsg->msgType != TDMT_MND_TELEM_TIMER) {
403 404 405 406 407 408 409 410
    SRpcConnInfo connInfo = {0};
    if ((pRpcMsg->msgType & 1U) && rpcGetConnInfo(pRpcMsg->handle, &connInfo) != 0) {
      taosFreeQitem(pMsg);
      terrno = TSDB_CODE_MND_NO_USER_FROM_CONN;
      mError("failed to create msg since %s, app:%p RPC:%p", terrstr(), pRpcMsg->ahandle, pRpcMsg->handle);
      return NULL;
    }
    memcpy(pMsg->user, connInfo.user, TSDB_USER_LEN);
S
Shengliang Guan 已提交
411 412
  }

S
Shengliang Guan 已提交
413
  pMsg->pMnode = pMnode;
S
Shengliang Guan 已提交
414 415 416
  pMsg->rpcMsg = *pRpcMsg;
  pMsg->createdTime = taosGetTimestampSec();

S
Shengliang Guan 已提交
417 418 419
  if (pRpcMsg != NULL) {
    mTrace("msg:%p, is created, app:%p RPC:%p user:%s", pMsg, pRpcMsg->ahandle, pRpcMsg->handle, pMsg->user);
  }
S
Shengliang Guan 已提交
420 421 422
  return pMsg;
}

423
void mndCleanupMsg(SMnodeMsg *pMsg) {
S
Shengliang Guan 已提交
424
  mTrace("msg:%p, is destroyed", pMsg);
S
Shengliang Guan 已提交
425 426
  rpcFreeCont(pMsg->rpcMsg.pCont);
  pMsg->rpcMsg.pCont = NULL;
S
Shengliang Guan 已提交
427 428 429
  taosFreeQitem(pMsg);
}

430 431 432 433
void mndSendRsp(SMnodeMsg *pMsg, int32_t code) {
  SRpcMsg rpcRsp = {.handle = pMsg->rpcMsg.handle, .code = code};
  rpcSendResponse(&rpcRsp);
}
S
Shengliang Guan 已提交
434

S
Shengliang Guan 已提交
435
void mndProcessMsg(SMnodeMsg *pMsg) {
436
  SMnode *pMnode = pMsg->pMnode;
S
Shengliang Guan 已提交
437
  int32_t code = 0;
438
  tmsg_t  msgType = pMsg->rpcMsg.msgType;
S
Shengliang Guan 已提交
439
  void   *ahandle = pMsg->rpcMsg.ahandle;
S
Shengliang Guan 已提交
440
  bool    isReq = (msgType & 1U);
441

S
Shengliang Guan 已提交
442
  mTrace("msg:%p, type:%s will be processed, app:%p", pMsg, TMSG_INFO(msgType), ahandle);
S
Shengliang Guan 已提交
443

S
Shengliang Guan 已提交
444 445
  if (isReq && !mndIsMaster(pMnode)) {
    code = TSDB_CODE_APP_NOT_READY;
S
Shengliang Guan 已提交
446
    mDebug("msg:%p, failed to process since %s, app:%p", pMsg, terrstr(), ahandle);
S
Shengliang Guan 已提交
447 448 449 450 451
    goto PROCESS_RPC_END;
  }

  if (isReq && pMsg->rpcMsg.pCont == NULL) {
    code = TSDB_CODE_MND_INVALID_MSG_LEN;
S
Shengliang Guan 已提交
452
    mError("msg:%p, failed to process since %s, app:%p", pMsg, terrstr(), ahandle);
S
Shengliang Guan 已提交
453
    goto PROCESS_RPC_END;
S
Shengliang Guan 已提交
454 455
  }

456
  MndMsgFp fp = pMnode->msgFp[TMSG_INDEX(msgType)];
S
Shengliang Guan 已提交
457
  if (fp == NULL) {
S
Shengliang Guan 已提交
458
    code = TSDB_CODE_MSG_NOT_PROCESSED;
S
Shengliang Guan 已提交
459
    mError("msg:%p, failed to process since no msg handle, app:%p", pMsg, ahandle);
S
Shengliang Guan 已提交
460
    goto PROCESS_RPC_END;
S
Shengliang Guan 已提交
461 462
  }

463
  code = (*fp)(pMsg);
S
Shengliang Guan 已提交
464
  if (code == TSDB_CODE_MND_ACTION_IN_PROGRESS) {
S
Shengliang Guan 已提交
465
    mTrace("msg:%p, in progress, app:%p", pMsg, ahandle);
S
Shengliang Guan 已提交
466 467
    return;
  } else if (code != 0) {
S
Shengliang Guan 已提交
468
    code = terrno;
S
Shengliang Guan 已提交
469
    mError("msg:%p, failed to process since %s, app:%p", pMsg, terrstr(), ahandle);
S
Shengliang Guan 已提交
470
    goto PROCESS_RPC_END;
471
  } else {
S
Shengliang Guan 已提交
472
    mTrace("msg:%p, is processed, app:%p", pMsg, ahandle);
S
Shengliang Guan 已提交
473 474 475 476
  }

PROCESS_RPC_END:
  if (isReq) {
S
Shengliang Guan 已提交
477 478
    if (pMsg->rpcMsg.handle == NULL) return;

S
Shengliang Guan 已提交
479
    if (code == TSDB_CODE_APP_NOT_READY) {
S
Shengliang Guan 已提交
480
      mndSendRedirectRsp(pMnode, &pMsg->rpcMsg);
S
Shengliang Guan 已提交
481
    } else if (code != 0) {
D
dapan1121 已提交
482
      SRpcMsg rpcRsp = {.handle = pMsg->rpcMsg.handle, .contLen = pMsg->contLen, .pCont = pMsg->pCont, .code = code};
S
Shengliang Guan 已提交
483
      rpcSendResponse(&rpcRsp);
S
Shengliang Guan 已提交
484
    } else {
S
Shengliang Guan 已提交
485 486
      SRpcMsg rpcRsp = {.handle = pMsg->rpcMsg.handle, .contLen = pMsg->contLen, .pCont = pMsg->pCont};
      rpcSendResponse(&rpcRsp);
S
Shengliang Guan 已提交
487
    }
S
Shengliang Guan 已提交
488
  }
S
Shengliang Guan 已提交
489 490
}

491 492 493 494
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 已提交
495 496 497
  }
}

D
dapan1121 已提交
498
// Note: uid 0 is reserved
S
Shengliang Guan 已提交
499
uint64_t mndGenerateUid(char *name, int32_t len) {
S
monitor  
Shengliang Guan 已提交
500
  int32_t hashval = MurmurHash3_32(name, len);
D
dapan1121 已提交
501 502 503 504 505 506 507 508 509

  do {
    int64_t  us = taosGetTimestampUs();
    uint64_t x = (us & 0x000000FFFFFFFFFF) << 24;
    uint64_t uuid = x + ((hashval & ((1ul << 16) - 1ul)) << 8) + (taosRand() & ((1ul << 8) - 1ul));
    if (uuid) {
      return uuid;
    }
  } while (true);
L
Liu Jicong 已提交
510
}
S
Shengliang Guan 已提交
511 512 513 514 515 516 517 518 519 520 521 522 523 524 525 526 527 528 529 530 531 532 533 534 535 536 537 538 539

void mndGetLoad(SMnode *pMnode, SMnodeLoad *pLoad) {
  memset(pLoad, 0, sizeof(SMnodeLoad));

  SSdb *pSdb = pMnode->pSdb;
  pLoad->numOfDnode = sdbGetSize(pSdb, SDB_DNODE);
  pLoad->numOfMnode = sdbGetSize(pSdb, SDB_MNODE);
  pLoad->numOfVgroup = sdbGetSize(pSdb, SDB_VGROUP);
  pLoad->numOfDatabase = sdbGetSize(pSdb, SDB_DB);
  pLoad->numOfSuperTable = sdbGetSize(pSdb, SDB_STB);

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

    pLoad->numOfChildTable += pVgroup->numOfTables;
    pLoad->numOfColumn += pVgroup->numOfTimeSeries;
    pLoad->totalPoints += pVgroup->pointsWritten;
    pLoad->totalStorage += pVgroup->totalStorage;
    pLoad->compStorage += pVgroup->compStorage;

    sdbRelease(pSdb, pVgroup);
  }
}

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

S
Shengliang Guan 已提交
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 567 568 569 570 571 572 573 574 575 576 577 578 579 580 581 582 583 584 585 586 587 588 589 590 591 592 593 594 595 596 597 598 599 600 601 602 603 604 605 606 607 608 609 610 611 612 613 614 615 616 617 618 619 620 621 622 623 624 625 626 627 628 629 630 631 632 633 634 635 636 637
  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));
    tstrncpy(desc.role, mndGetRoleStr(pObj->role), sizeof(desc.role));
    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;
    strncpy(desc.database_name, pVgroup->dbName, sizeof(desc.database_name));
    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;
      tstrncpy(pVnDesc->vnode_role, mndGetRoleStr(pVgid->role), sizeof(pVnDesc->vnode_role));
      if (pVgid->role == TAOS_SYNC_STATE_LEADER) {
        tstrncpy(desc.status, "ready", sizeof(desc.status));
        pClusterInfo->vgroups_alive++;
      }
      if (pVgid->role == TAOS_SYNC_STATE_LEADER || pVgid->role == TAOS_SYNC_STATE_CANDIDATE) {
        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 已提交
638
}