mndVgroup.c 21.1 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
#include "mndVgroup.h"
S
Shengliang Guan 已提交
18
#include "mndDb.h"
S
Shengliang Guan 已提交
19
#include "mndDnode.h"
S
Shengliang Guan 已提交
20
#include "mndMnode.h"
S
Shengliang Guan 已提交
21 22
#include "mndShow.h"
#include "mndTrans.h"
S
Shengliang Guan 已提交
23

L
Liu Jicong 已提交
24
#define TSDB_VGROUP_VER_NUMBER   1
S
Shengliang Guan 已提交
25 26
#define TSDB_VGROUP_RESERVE_SIZE 64

S
Shengliang Guan 已提交
27 28 29
static SSdbRow *mndVgroupActionDecode(SSdbRaw *pRaw);
static int32_t  mndVgroupActionInsert(SSdb *pSdb, SVgObj *pVgroup);
static int32_t  mndVgroupActionDelete(SSdb *pSdb, SVgObj *pVgroup);
S
Shengliang Guan 已提交
30
static int32_t  mndVgroupActionUpdate(SSdb *pSdb, SVgObj *pOld, SVgObj *pNew);
S
Shengliang Guan 已提交
31

S
Shengliang Guan 已提交
32 33 34 35 36
static int32_t mndProcessCreateVnodeRsp(SNodeMsg *pRsp);
static int32_t mndProcessAlterVnodeRsp(SNodeMsg *pRsp);
static int32_t mndProcessDropVnodeRsp(SNodeMsg *pRsp);
static int32_t mndProcessCompactVnodeRsp(SNodeMsg *pRsp);

37
static int32_t mndRetrieveVgroups(SNodeMsg *pReq, SShowObj *pShow, SSDataBlock* pBlock, int32_t rows);
S
Shengliang Guan 已提交
38
static void    mndCancelGetNextVgroup(SMnode *pMnode, void *pIter);
39
static int32_t mndRetrieveVnodes(SNodeMsg *pReq, SShowObj *pShow, SSDataBlock* pBlock, int32_t rows);
S
Shengliang Guan 已提交
40 41 42
static void    mndCancelGetNextVnode(SMnode *pMnode, void *pIter);

int32_t mndInitVgroup(SMnode *pMnode) {
S
Shengliang Guan 已提交
43
  SSdbTable table = {.sdbType = SDB_VGROUP,
44
                     .keyType = SDB_KEY_INT32,
S
Shengliang Guan 已提交
45 46 47 48 49 50
                     .encodeFp = (SdbEncodeFp)mndVgroupActionEncode,
                     .decodeFp = (SdbDecodeFp)mndVgroupActionDecode,
                     .insertFp = (SdbInsertFp)mndVgroupActionInsert,
                     .updateFp = (SdbUpdateFp)mndVgroupActionDelete,
                     .deleteFp = (SdbDeleteFp)mndVgroupActionUpdate};

H
Hongze Cheng 已提交
51
  mndSetMsgHandle(pMnode, TDMT_DND_CREATE_VNODE_RSP, mndProcessCreateVnodeRsp);
S
Shengliang Guan 已提交
52
  mndSetMsgHandle(pMnode, TDMT_VND_ALTER_VNODE_RSP, mndProcessAlterVnodeRsp);
H
Hongze Cheng 已提交
53
  mndSetMsgHandle(pMnode, TDMT_DND_DROP_VNODE_RSP, mndProcessDropVnodeRsp);
S
Shengliang Guan 已提交
54
  mndSetMsgHandle(pMnode, TDMT_VND_COMPACT_VNODE_RSP, mndProcessCompactVnodeRsp);
S
Shengliang Guan 已提交
55 56 57

  mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_VGROUP, mndRetrieveVgroups);
  mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_VGROUP, mndCancelGetNextVgroup);
S
Shengliang Guan 已提交
58 59 60
  mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_VNODES, mndRetrieveVnodes);
  mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_VNODES, mndCancelGetNextVnode);

S
Shengliang Guan 已提交
61
  return sdbSetTable(pMnode->pSdb, table);
S
Shengliang Guan 已提交
62 63 64 65
}

void mndCleanupVgroup(SMnode *pMnode) {}

S
Shengliang Guan 已提交
66
SSdbRaw *mndVgroupActionEncode(SVgObj *pVgroup) {
67 68
  terrno = TSDB_CODE_OUT_OF_MEMORY;

S
Shengliang Guan 已提交
69
  SSdbRaw *pRaw = sdbAllocRaw(SDB_VGROUP, TSDB_VGROUP_VER_NUMBER, sizeof(SVgObj) + TSDB_VGROUP_RESERVE_SIZE);
70
  if (pRaw == NULL) goto VG_ENCODE_OVER;
S
Shengliang Guan 已提交
71 72

  int32_t dataPos = 0;
73 74 75 76 77 78 79 80 81
  SDB_SET_INT32(pRaw, dataPos, pVgroup->vgId, VG_ENCODE_OVER)
  SDB_SET_INT64(pRaw, dataPos, pVgroup->createdTime, VG_ENCODE_OVER)
  SDB_SET_INT64(pRaw, dataPos, pVgroup->updateTime, VG_ENCODE_OVER)
  SDB_SET_INT32(pRaw, dataPos, pVgroup->version, VG_ENCODE_OVER)
  SDB_SET_INT32(pRaw, dataPos, pVgroup->hashBegin, VG_ENCODE_OVER)
  SDB_SET_INT32(pRaw, dataPos, pVgroup->hashEnd, VG_ENCODE_OVER)
  SDB_SET_BINARY(pRaw, dataPos, pVgroup->dbName, TSDB_DB_FNAME_LEN, VG_ENCODE_OVER)
  SDB_SET_INT64(pRaw, dataPos, pVgroup->dbUid, VG_ENCODE_OVER)
  SDB_SET_INT8(pRaw, dataPos, pVgroup->replica, VG_ENCODE_OVER)
S
Shengliang Guan 已提交
82 83
  for (int8_t i = 0; i < pVgroup->replica; ++i) {
    SVnodeGid *pVgid = &pVgroup->vnodeGid[i];
84 85 86 87 88 89 90 91 92 93 94 95
    SDB_SET_INT32(pRaw, dataPos, pVgid->dnodeId, VG_ENCODE_OVER)
  }
  SDB_SET_RESERVE(pRaw, dataPos, TSDB_VGROUP_RESERVE_SIZE, VG_ENCODE_OVER)
  SDB_SET_DATALEN(pRaw, dataPos, VG_ENCODE_OVER)

  terrno = 0;

VG_ENCODE_OVER:
  if (terrno != 0) {
    mError("vgId:%d, failed to encode to raw:%p since %s", pVgroup->vgId, pRaw, terrstr());
    sdbFreeRaw(pRaw);
    return NULL;
S
Shengliang Guan 已提交
96 97
  }

98
  mTrace("vgId:%d, encode to raw:%p, row:%p", pVgroup->vgId, pRaw, pVgroup);
S
Shengliang Guan 已提交
99 100 101
  return pRaw;
}

S
Shengliang Guan 已提交
102
SSdbRow *mndVgroupActionDecode(SSdbRaw *pRaw) {
103 104
  terrno = TSDB_CODE_OUT_OF_MEMORY;

S
Shengliang Guan 已提交
105
  int8_t sver = 0;
106
  if (sdbGetRawSoftVer(pRaw, &sver) != 0) goto VG_DECODE_OVER;
S
Shengliang Guan 已提交
107

S
Shengliang Guan 已提交
108
  if (sver != TSDB_VGROUP_VER_NUMBER) {
S
Shengliang Guan 已提交
109
    terrno = TSDB_CODE_SDB_INVALID_DATA_VER;
110
    goto VG_DECODE_OVER;
S
Shengliang Guan 已提交
111 112
  }

S
Shengliang Guan 已提交
113
  SSdbRow *pRow = sdbAllocRow(sizeof(SVgObj));
114 115 116 117
  if (pRow == NULL) goto VG_DECODE_OVER;

  SVgObj *pVgroup = sdbGetRowObj(pRow);
  if (pVgroup == NULL) goto VG_DECODE_OVER;
S
Shengliang Guan 已提交
118 119

  int32_t dataPos = 0;
120 121 122 123 124 125 126 127 128
  SDB_GET_INT32(pRaw, dataPos, &pVgroup->vgId, VG_DECODE_OVER)
  SDB_GET_INT64(pRaw, dataPos, &pVgroup->createdTime, VG_DECODE_OVER)
  SDB_GET_INT64(pRaw, dataPos, &pVgroup->updateTime, VG_DECODE_OVER)
  SDB_GET_INT32(pRaw, dataPos, &pVgroup->version, VG_DECODE_OVER)
  SDB_GET_INT32(pRaw, dataPos, &pVgroup->hashBegin, VG_DECODE_OVER)
  SDB_GET_INT32(pRaw, dataPos, &pVgroup->hashEnd, VG_DECODE_OVER)
  SDB_GET_BINARY(pRaw, dataPos, pVgroup->dbName, TSDB_DB_FNAME_LEN, VG_DECODE_OVER)
  SDB_GET_INT64(pRaw, dataPos, &pVgroup->dbUid, VG_DECODE_OVER)
  SDB_GET_INT8(pRaw, dataPos, &pVgroup->replica, VG_DECODE_OVER)
S
Shengliang Guan 已提交
129 130
  for (int8_t i = 0; i < pVgroup->replica; ++i) {
    SVnodeGid *pVgid = &pVgroup->vnodeGid[i];
131
    SDB_GET_INT32(pRaw, dataPos, &pVgid->dnodeId, VG_DECODE_OVER)
S
Shengliang Guan 已提交
132 133 134
    if (pVgroup->replica == 1) {
      pVgid->role = TAOS_SYNC_STATE_LEADER;
    }
S
Shengliang Guan 已提交
135
  }
136 137 138 139 140 141 142
  SDB_GET_RESERVE(pRaw, dataPos, TSDB_VGROUP_RESERVE_SIZE, VG_DECODE_OVER)

  terrno = 0;

VG_DECODE_OVER:
  if (terrno != 0) {
    mError("vgId:%d, failed to decode from raw:%p since %s", pVgroup->vgId, pRaw, terrstr());
wafwerar's avatar
wafwerar 已提交
143
    taosMemoryFreeClear(pRow);
144 145
    return NULL;
  }
S
Shengliang Guan 已提交
146

147
  mTrace("vgId:%d, decode from raw:%p, row:%p", pVgroup->vgId, pRaw, pVgroup);
S
Shengliang Guan 已提交
148 149 150 151
  return pRow;
}

static int32_t mndVgroupActionInsert(SSdb *pSdb, SVgObj *pVgroup) {
152
  mTrace("vgId:%d, perform insert action, row:%p", pVgroup->vgId, pVgroup);
S
Shengliang Guan 已提交
153 154 155 156
  return 0;
}

static int32_t mndVgroupActionDelete(SSdb *pSdb, SVgObj *pVgroup) {
157
  mTrace("vgId:%d, perform delete action, row:%p", pVgroup->vgId, pVgroup);
S
Shengliang Guan 已提交
158 159 160
  return 0;
}

S
Shengliang Guan 已提交
161 162 163 164 165 166 167 168
static int32_t mndVgroupActionUpdate(SSdb *pSdb, SVgObj *pOld, SVgObj *pNew) {
  mTrace("vgId:%d, perform update action, old row:%p new row:%p", pOld->vgId, pOld, pNew);
  pOld->updateTime = pNew->updateTime;
  pOld->version = pNew->version;
  pOld->hashBegin = pNew->hashBegin;
  pOld->hashEnd = pNew->hashEnd;
  pOld->replica = pNew->replica;
  memcpy(pOld->vnodeGid, pNew->vnodeGid, TSDB_MAX_REPLICA * sizeof(SVnodeGid));
S
Shengliang Guan 已提交
169 170 171 172
  return 0;
}

SVgObj *mndAcquireVgroup(SMnode *pMnode, int32_t vgId) {
S
Shengliang Guan 已提交
173 174
  SSdb   *pSdb = pMnode->pSdb;
  SVgObj *pVgroup = sdbAcquire(pSdb, SDB_VGROUP, &vgId);
S
Shengliang Guan 已提交
175
  if (pVgroup == NULL && terrno == TSDB_CODE_SDB_OBJ_NOT_THERE) {
S
Shengliang Guan 已提交
176 177 178
    terrno = TSDB_CODE_MND_VGROUP_NOT_EXIST;
  }
  return pVgroup;
S
Shengliang Guan 已提交
179 180 181 182 183 184 185
}

void mndReleaseVgroup(SMnode *pMnode, SVgObj *pVgroup) {
  SSdb *pSdb = pMnode->pSdb;
  sdbRelease(pSdb, pVgroup);
}

S
Shengliang Guan 已提交
186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205
void *mndBuildCreateVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup, int32_t *pContLen) {
  SCreateVnodeReq createReq = {0};
  createReq.vgId = pVgroup->vgId;
  createReq.dnodeId = pDnode->id;
  memcpy(createReq.db, pDb->name, TSDB_DB_FNAME_LEN);
  createReq.dbUid = pDb->uid;
  createReq.vgVersion = pVgroup->version;
  createReq.cacheBlockSize = pDb->cfg.cacheBlockSize;
  createReq.totalBlocks = pDb->cfg.totalBlocks;
  createReq.daysPerFile = pDb->cfg.daysPerFile;
  createReq.daysToKeep0 = pDb->cfg.daysToKeep0;
  createReq.daysToKeep1 = pDb->cfg.daysToKeep1;
  createReq.daysToKeep2 = pDb->cfg.daysToKeep2;
  createReq.minRows = pDb->cfg.minRows;
  createReq.maxRows = pDb->cfg.maxRows;
  createReq.commitTime = pDb->cfg.commitTime;
  createReq.fsyncPeriod = pDb->cfg.fsyncPeriod;
  createReq.walLevel = pDb->cfg.walLevel;
  createReq.precision = pDb->cfg.precision;
  createReq.compression = pDb->cfg.compression;
S
Shengliang Guan 已提交
206
  createReq.strict = pDb->cfg.strict;
S
Shengliang Guan 已提交
207 208 209 210
  createReq.update = pDb->cfg.update;
  createReq.cacheLastRow = pDb->cfg.cacheLastRow;
  createReq.replica = pVgroup->replica;
  createReq.selfIndex = -1;
L
Liu Jicong 已提交
211
  createReq.streamMode = pVgroup->streamMode;
D
dapan1121 已提交
212 213
  createReq.hashBegin = pVgroup->hashBegin;
  createReq.hashEnd = pVgroup->hashEnd;
S
Shengliang Guan 已提交
214
  createReq.hashMethod = pDb->cfg.hashMethod;
S
sma  
Shengliang Guan 已提交
215 216
  createReq.numOfRetensions = pDb->cfg.numOfRetensions;
  createReq.pRetensions = pDb->cfg.pRetensions;
S
Shengliang Guan 已提交
217 218

  for (int32_t v = 0; v < pVgroup->replica; ++v) {
S
Shengliang Guan 已提交
219
    SReplica  *pReplica = &createReq.replicas[v];
S
Shengliang Guan 已提交
220 221 222 223 224 225
    SVnodeGid *pVgid = &pVgroup->vnodeGid[v];
    SDnodeObj *pVgidDnode = mndAcquireDnode(pMnode, pVgid->dnodeId);
    if (pVgidDnode == NULL) {
      return NULL;
    }

S
Shengliang Guan 已提交
226 227
    pReplica->id = pVgidDnode->id;
    pReplica->port = pVgidDnode->port;
S
Shengliang Guan 已提交
228 229 230 231
    memcpy(pReplica->fqdn, pVgidDnode->fqdn, TSDB_FQDN_LEN);
    mndReleaseDnode(pMnode, pVgidDnode);

    if (pDnode->id == pVgid->dnodeId) {
S
Shengliang Guan 已提交
232
      createReq.selfIndex = v;
S
Shengliang Guan 已提交
233 234 235
    }
  }

S
Shengliang Guan 已提交
236
  if (createReq.selfIndex == -1) {
S
Shengliang Guan 已提交
237 238 239 240
    terrno = TSDB_CODE_MND_APP_ERROR;
    return NULL;
  }

S
Shengliang Guan 已提交
241 242 243 244 245 246
  int32_t contLen = tSerializeSCreateVnodeReq(NULL, 0, &createReq);
  if (contLen < 0) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return NULL;
  }

wafwerar's avatar
wafwerar 已提交
247
  void *pReq = taosMemoryMalloc(contLen);
S
Shengliang Guan 已提交
248 249 250 251 252 253 254 255
  if (pReq == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return NULL;
  }

  tSerializeSCreateVnodeReq(pReq, contLen, &createReq);
  *pContLen = contLen;
  return pReq;
S
Shengliang Guan 已提交
256 257
}

L
Liu Jicong 已提交
258
void *mndBuildDropVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup, int32_t *pContLen) {
S
Shengliang Guan 已提交
259 260 261 262 263 264 265 266
  SDropVnodeReq dropReq = {0};
  dropReq.dnodeId = pDnode->id;
  dropReq.vgId = pVgroup->vgId;
  memcpy(dropReq.db, pDb->name, TSDB_DB_FNAME_LEN);
  dropReq.dbUid = pDb->uid;

  int32_t contLen = tSerializeSDropVnodeReq(NULL, 0, &dropReq);
  if (contLen < 0) {
S
Shengliang Guan 已提交
267 268 269 270
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return NULL;
  }

wafwerar's avatar
wafwerar 已提交
271
  void *pReq = taosMemoryMalloc(contLen);
S
Shengliang Guan 已提交
272 273 274 275
  if (pReq == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return NULL;
  }
S
Shengliang Guan 已提交
276

S
Shengliang Guan 已提交
277 278 279
  tSerializeSDropVnodeReq(pReq, contLen, &dropReq);
  *pContLen = contLen;
  return pReq;
S
Shengliang Guan 已提交
280 281
}

S
Shengliang Guan 已提交
282 283 284 285 286 287 288 289 290 291 292 293
static bool mndResetDnodesArrayFp(SMnode *pMnode, void *pObj, void *p1, void *p2, void *p3) {
  SDnodeObj *pDnode = pObj;
  pDnode->numOfVnodes = 0;
  return true;
}

static bool mndBuildDnodesArrayFp(SMnode *pMnode, void *pObj, void *p1, void *p2, void *p3) {
  SDnodeObj *pDnode = pObj;
  SArray    *pArray = p1;

  int64_t curMs = taosGetTimestampMs();
  bool    online = mndIsDnodeOnline(pMnode, pDnode, curMs);
S
Shengliang Guan 已提交
294 295
  bool    isMnode = mndIsMnode(pMnode, pDnode->id);
  pDnode->numOfVnodes = mndGetVnodesNum(pMnode, pDnode->id);
S
Shengliang Guan 已提交
296 297 298 299 300 301 302 303

  mDebug("dnode:%d, vnodes:%d supportVnodes:%d isMnode:%d online:%d", pDnode->id, pDnode->numOfVnodes,
         pDnode->numOfSupportVnodes, isMnode, online);

  if (isMnode) {
    pDnode->numOfVnodes++;
  }

S
Shengliang Guan 已提交
304 305 306
  if (online && pDnode->numOfSupportVnodes > 0) {
    taosArrayPush(pArray, pDnode);
  }
S
Shengliang Guan 已提交
307 308 309
  return true;
}

S
Shengliang Guan 已提交
310
static SArray *mndBuildDnodesArray(SMnode *pMnode) {
S
Shengliang Guan 已提交
311
  SSdb   *pSdb = pMnode->pSdb;
S
Shengliang Guan 已提交
312
  int32_t numOfDnodes = mndGetDnodeSize(pMnode);
S
Shengliang Guan 已提交
313

S
Shengliang Guan 已提交
314 315 316 317 318
  SArray *pArray = taosArrayInit(numOfDnodes, sizeof(SDnodeObj));
  if (pArray == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return NULL;
  }
S
Shengliang Guan 已提交
319

S
Shengliang Guan 已提交
320 321
  sdbTraverse(pSdb, SDB_DNODE, mndResetDnodesArrayFp, NULL, NULL, NULL);
  sdbTraverse(pSdb, SDB_DNODE, mndBuildDnodesArrayFp, pArray, NULL, NULL);
S
Shengliang Guan 已提交
322 323 324 325 326 327
  return pArray;
}

static int32_t mndCompareDnodeVnodes(SDnodeObj *pDnode1, SDnodeObj *pDnode2) {
  float d1Score = (float)pDnode1->numOfVnodes / pDnode1->numOfSupportVnodes;
  float d2Score = (float)pDnode2->numOfVnodes / pDnode2->numOfSupportVnodes;
S
Shengliang Guan 已提交
328
  return d1Score >= d2Score ? 1 : 0;
S
Shengliang Guan 已提交
329 330 331 332 333 334 335 336 337
}

static int32_t mndGetAvailableDnode(SMnode *pMnode, SVgObj *pVgroup, SArray *pArray) {
  SSdb   *pSdb = pMnode->pSdb;
  int32_t allocedVnodes = 0;
  void   *pIter = NULL;

  taosArraySort(pArray, (__compar_fn_t)mndCompareDnodeVnodes);

S
Shengliang Guan 已提交
338 339 340 341 342 343 344 345
  int32_t size = taosArrayGetSize(pArray);
  if (size < pVgroup->replica) {
    mError("db:%s, vgId:%d, no enough online dnodes:%d to alloc %d replica", pVgroup->dbName, pVgroup->vgId, size,
           pVgroup->replica);
    terrno = TSDB_CODE_MND_NO_ENOUGH_DNODES;
    return -1;
  }

S
Shengliang Guan 已提交
346 347 348 349 350 351 352 353 354 355 356 357 358 359 360
  for (int32_t v = 0; v < pVgroup->replica; ++v) {
    SVnodeGid *pVgid = &pVgroup->vnodeGid[v];
    SDnodeObj *pDnode = taosArrayGet(pArray, v);
    if (pDnode == NULL || pDnode->numOfVnodes > pDnode->numOfSupportVnodes) {
      terrno = TSDB_CODE_MND_NO_ENOUGH_DNODES;
      return -1;
    }

    pVgid->dnodeId = pDnode->id;
    if (pVgroup->replica == 1) {
      pVgid->role = TAOS_SYNC_STATE_LEADER;
    } else {
      pVgid->role = TAOS_SYNC_STATE_FOLLOWER;
    }

S
Shengliang Guan 已提交
361
    mDebug("db:%s, vgId:%d, vn:%d dnode:%d is alloced", pVgroup->dbName, pVgroup->vgId, v, pVgid->dnodeId);
S
Shengliang Guan 已提交
362
    pDnode->numOfVnodes++;
S
Shengliang Guan 已提交
363
  }
S
Shengliang Guan 已提交
364

S
Shengliang Guan 已提交
365 366
  return 0;
}
367

S
Shengliang Guan 已提交
368
int32_t mndAllocVgroup(SMnode *pMnode, SDbObj *pDb, SVgObj **ppVgroups) {
S
Shengliang Guan 已提交
369 370 371 372
  int32_t code = -1;
  SArray *pArray = NULL;
  SVgObj *pVgroups = NULL;

wafwerar's avatar
wafwerar 已提交
373
  pVgroups = taosMemoryCalloc(pDb->cfg.numOfVgroups, sizeof(SVgObj));
S
Shengliang Guan 已提交
374 375
  if (pVgroups == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
S
Shengliang Guan 已提交
376
    goto ALLOC_VGROUP_OVER;
S
Shengliang Guan 已提交
377 378
  }

S
Shengliang Guan 已提交
379 380 381 382 383 384 385 386
  pArray = mndBuildDnodesArray(pMnode);
  if (pArray == NULL) {
    goto ALLOC_VGROUP_OVER;
  }

  mDebug("db:%s, total %d dnodes used to create %d vgroups (%d vnodes)", pDb->name, (int32_t)taosArrayGetSize(pArray),
         pDb->cfg.numOfVgroups, pDb->cfg.numOfVgroups * pDb->cfg.replications);

S
Shengliang Guan 已提交
387
  int32_t  allocedVgroups = 0;
S
Shengliang Guan 已提交
388 389 390
  int32_t  maxVgId = sdbGetMaxId(pMnode->pSdb, SDB_VGROUP);
  uint32_t hashMin = 0;
  uint32_t hashMax = UINT32_MAX;
391
  uint32_t hashInterval = (hashMax - hashMin) / pDb->cfg.numOfVgroups;
S
Shengliang Guan 已提交
392

393 394
  if (maxVgId < 2) maxVgId = 2;

395
  for (uint32_t v = 0; v < pDb->cfg.numOfVgroups; v++) {
S
Shengliang Guan 已提交
396
    SVgObj *pVgroup = &pVgroups[v];
S
Shengliang Guan 已提交
397
    pVgroup->vgId = maxVgId++;
S
Shengliang Guan 已提交
398 399
    pVgroup->createdTime = taosGetTimestampMs();
    pVgroup->updateTime = pVgroups->createdTime;
S
Shengliang Guan 已提交
400
    pVgroup->version = 1;
L
Liu Jicong 已提交
401
    pVgroup->streamMode = pDb->cfg.streamMode;
S
Shengliang Guan 已提交
402
    pVgroup->hashBegin = hashMin + hashInterval * v;
403
    if (v == pDb->cfg.numOfVgroups - 1) {
S
Shengliang Guan 已提交
404 405
      pVgroup->hashEnd = hashMax;
    } else {
S
Shengliang Guan 已提交
406
      pVgroup->hashEnd = hashMin + hashInterval * (v + 1) - 1;
S
Shengliang Guan 已提交
407
    }
S
Shengliang Guan 已提交
408

409
    memcpy(pVgroup->dbName, pDb->name, TSDB_DB_FNAME_LEN);
S
Shengliang Guan 已提交
410
    pVgroup->dbUid = pDb->uid;
S
Shengliang Guan 已提交
411
    pVgroup->replica = pDb->cfg.replications;
S
Shengliang Guan 已提交
412

S
Shengliang Guan 已提交
413
    if (mndGetAvailableDnode(pMnode, pVgroup, pArray) != 0) {
S
Shengliang Guan 已提交
414
      terrno = TSDB_CODE_MND_NO_ENOUGH_DNODES;
S
Shengliang Guan 已提交
415
      goto ALLOC_VGROUP_OVER;
S
Shengliang Guan 已提交
416 417
    }

S
Shengliang Guan 已提交
418
    allocedVgroups++;
419 420
  }

S
Shengliang Guan 已提交
421
  *ppVgroups = pVgroups;
S
Shengliang Guan 已提交
422 423 424 425 426
  code = 0;

  mDebug("db:%s, %d vgroups is alloced, replica:%d", pDb->name, pDb->cfg.numOfVgroups, pDb->cfg.replications);

ALLOC_VGROUP_OVER:
wafwerar's avatar
wafwerar 已提交
427
  if (code != 0) taosMemoryFree(pVgroups);
S
Shengliang Guan 已提交
428 429
  taosArrayDestroy(pArray);
  return code;
430 431
}

L
Liu Jicong 已提交
432
SEpSet mndGetVgroupEpset(SMnode *pMnode, const SVgObj *pVgroup) {
433 434 435
  SEpSet epset = {0};

  for (int32_t v = 0; v < pVgroup->replica; ++v) {
L
Liu Jicong 已提交
436 437
    const SVnodeGid *pVgid = &pVgroup->vnodeGid[v];
    SDnodeObj       *pDnode = mndAcquireDnode(pMnode, pVgid->dnodeId);
438 439 440 441 442 443
    if (pDnode == NULL) continue;

    if (pVgid->role == TAOS_SYNC_STATE_LEADER) {
      epset.inUse = epset.numOfEps;
    }

H
Haojun Liao 已提交
444
    addEpIntoEpSet(&epset, pDnode->fqdn, pDnode->port);
445 446 447 448 449 450
    mndReleaseDnode(pMnode, pDnode);
  }

  return epset;
}

S
Shengliang Guan 已提交
451
static int32_t mndProcessCreateVnodeRsp(SNodeMsg *pRsp) {
S
Shengliang Guan 已提交
452
  mndTransProcessRsp(pRsp);
453 454 455
  return 0;
}

S
Shengliang Guan 已提交
456
static int32_t mndProcessAlterVnodeRsp(SNodeMsg *pRsp) {
S
Shengliang Guan 已提交
457
  mndTransProcessRsp(pRsp);
S
Shengliang Guan 已提交
458 459 460
  return 0;
}

S
Shengliang Guan 已提交
461
static int32_t mndProcessDropVnodeRsp(SNodeMsg *pRsp) {
S
Shengliang Guan 已提交
462
  mndTransProcessRsp(pRsp);
S
Shengliang Guan 已提交
463 464 465
  return 0;
}

S
Shengliang Guan 已提交
466
static int32_t mndProcessCompactVnodeRsp(SNodeMsg *pRsp) { return 0; }
S
Shengliang Guan 已提交
467

S
Shengliang Guan 已提交
468 469 470 471 472 473 474
static bool mndGetVgroupMaxReplicaFp(SMnode *pMnode, void *pObj, void *p1, void *p2, void *p3) {
  SVgObj  *pVgroup = pObj;
  int64_t  uid = *(int64_t *)p1;
  int8_t  *pReplica = p2;
  int32_t *pNumOfVgroups = p3;

  if (pVgroup->dbUid == uid) {
dengyihao's avatar
dengyihao 已提交
475
    *pReplica = TMAX(*pReplica, pVgroup->replica);
S
Shengliang Guan 已提交
476 477 478 479 480 481
    (*pNumOfVgroups)++;
  }

  return true;
}

S
Shengliang Guan 已提交
482
static int32_t mndGetVgroupMaxReplica(SMnode *pMnode, char *dbName, int8_t *pReplica, int32_t *pNumOfVgroups) {
S
Shengliang Guan 已提交
483
  SSdb   *pSdb = pMnode->pSdb;
S
Shengliang Guan 已提交
484 485 486 487 488 489
  SDbObj *pDb = mndAcquireDb(pMnode, dbName);
  if (pDb == NULL) {
    terrno = TSDB_CODE_MND_DB_NOT_SELECTED;
    return -1;
  }

S
Shengliang Guan 已提交
490 491 492 493
  *pReplica = 1;
  *pNumOfVgroups = 0;
  sdbTraverse(pSdb, SDB_VGROUP, mndGetVgroupMaxReplicaFp, &pDb->uid, pReplica, pNumOfVgroups);
  mndReleaseDb(pMnode, pDb);
S
Shengliang Guan 已提交
494
  return 0;
S
Shengliang Guan 已提交
495 496
}

497
static int32_t mndRetrieveVgroups(SNodeMsg *pReq, SShowObj *pShow, SSDataBlock* pBlock, int32_t rows) {
S
Shengliang Guan 已提交
498
  SMnode *pMnode = pReq->pNode;
S
Shengliang Guan 已提交
499 500 501 502 503
  SSdb   *pSdb = pMnode->pSdb;
  int32_t numOfRows = 0;
  SVgObj *pVgroup = NULL;
  int32_t cols = 0;

H
Haojun Liao 已提交
504 505 506 507 508 509 510
  SDbObj *pDb = NULL;
  if (strlen(pShow->db) > 0) {
    pDb = mndAcquireDb(pMnode, pShow->db);
    if (pDb == NULL) {
      return 0;
    }
  }
S
Shengliang Guan 已提交
511

S
Shengliang Guan 已提交
512 513 514 515
  while (numOfRows < rows) {
    pShow->pIter = sdbFetch(pSdb, SDB_VGROUP, pShow->pIter, (void **)&pVgroup);
    if (pShow->pIter == NULL) break;

H
Haojun Liao 已提交
516 517 518 519 520
    if (pDb != NULL && pVgroup->dbUid != pDb->uid) {
      continue;
    }

    cols = 0;
521 522
    SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    colDataAppend(pColInfo, numOfRows, (const char *)&pVgroup->vgId, false);
H
Haojun Liao 已提交
523 524

    SName name = {0};
525 526 527 528
    char  db[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
    tNameFromString(&name, pVgroup->dbName, T_NAME_ACCT | T_NAME_DB);
    tNameGetDbName(&name, varDataVal(db));
    varDataSetLen(db, strlen(varDataVal(db)));
H
Haojun Liao 已提交
529

530 531
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    colDataAppend(pColInfo, numOfRows, (const char *)db, false);
H
Haojun Liao 已提交
532

533 534
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    colDataAppend(pColInfo, numOfRows, (const char *)&pVgroup->numOfTables, false);
H
Haojun Liao 已提交
535

536 537
    // default 3 replica
    for (int32_t i = 0; i < 3; ++i) {
538 539

      pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
540 541 542 543
      if (i < pVgroup->replica) {
        colDataAppend(pColInfo, numOfRows, (const char *)&pVgroup->vnodeGid[i].dnodeId, false);

        char        buf1[20] = {0};
S
Shengliang Guan 已提交
544
        const char *role = syncStr(pVgroup->vnodeGid[i].role);
545 546 547 548 549 550 551 552 553
        STR_WITH_MAXSIZE_TO_VARSTR(buf1, role, pShow->bytes[cols]);

        pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
        colDataAppend(pColInfo, numOfRows, (const char *)buf1, false);
      } else {
        colDataAppendNULL(pColInfo, numOfRows);
        pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
        colDataAppendNULL(pColInfo, numOfRows);
      }
S
Shengliang Guan 已提交
554
    }
S
Shengliang Guan 已提交
555

556 557 558 559 560 561 562 563 564
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    colDataAppendNULL(pColInfo, numOfRows);

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    colDataAppendNULL(pColInfo, numOfRows);

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols);
    colDataAppendNULL(pColInfo, numOfRows);

H
Haojun Liao 已提交
565
    numOfRows++;
S
Shengliang Guan 已提交
566 567 568
    sdbRelease(pSdb, pVgroup);
  }

H
Haojun Liao 已提交
569 570 571 572
  if (pDb != NULL) {
    mndReleaseDb(pMnode, pDb);
  }

573
  pShow->numOfRows += numOfRows;
S
Shengliang Guan 已提交
574 575 576 577 578 579 580 581
  return numOfRows;
}

static void mndCancelGetNextVgroup(SMnode *pMnode, void *pIter) {
  SSdb *pSdb = pMnode->pSdb;
  sdbCancelFetch(pSdb, pIter);
}

S
Shengliang Guan 已提交
582 583 584 585
static bool mndGetVnodesNumFp(SMnode *pMnode, void *pObj, void *p1, void *p2, void *p3) {
  SVgObj  *pVgroup = pObj;
  int32_t  dnodeId = *(int32_t *)p1;
  int32_t *pNumOfVnodes = (int32_t *)p2;
S
Shengliang Guan 已提交
586

S
Shengliang Guan 已提交
587 588 589
  for (int32_t v = 0; v < pVgroup->replica; ++v) {
    if (pVgroup->vnodeGid[v].dnodeId == dnodeId) {
      (*pNumOfVnodes)++;
S
Shengliang Guan 已提交
590
    }
S
Shengliang Guan 已提交
591 592
  }

S
Shengliang Guan 已提交
593 594 595 596 597 598
  return true;
}

int32_t mndGetVnodesNum(SMnode *pMnode, int32_t dnodeId) {
  int32_t numOfVnodes = 0;
  sdbTraverse(pMnode->pSdb, SDB_VGROUP, mndGetVnodesNumFp, &dnodeId, &numOfVnodes, NULL);
S
Shengliang Guan 已提交
599
  return numOfVnodes;
S
Shengliang Guan 已提交
600 601
}

602
static int32_t mndRetrieveVnodes(SNodeMsg *pReq, SShowObj *pShow, SSDataBlock* pBlock, int32_t rows) {
S
Shengliang Guan 已提交
603
  SMnode *pMnode = pReq->pNode;
S
Shengliang Guan 已提交
604 605 606 607
  SSdb   *pSdb = pMnode->pSdb;
  int32_t numOfRows = 0;
  SVgObj *pVgroup = NULL;
  int32_t cols = 0;
H
Haojun Liao 已提交
608
//  int32_t dnodeId = pShow->replica;
S
Shengliang Guan 已提交
609 610 611 612 613

  while (numOfRows < rows) {
    pShow->pIter = sdbFetch(pSdb, SDB_VGROUP, pShow->pIter, (void **)&pVgroup);
    if (pShow->pIter == NULL) break;

S
Shengliang Guan 已提交
614
    for (int32_t i = 0; i < pVgroup->replica && numOfRows < rows; ++i) {
S
Shengliang Guan 已提交
615 616 617
      SVnodeGid *pVgid = &pVgroup->vnodeGid[i];
      cols = 0;

618 619
      SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
      colDataAppend(pColInfo, numOfRows, (const char *)&pVgroup->vgId, false);
S
Shengliang Guan 已提交
620

H
Haojun Liao 已提交
621
      SName name = {0};
622 623 624 625
      char  db[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
      tNameFromString(&name, pVgroup->dbName, T_NAME_ACCT | T_NAME_DB);
      tNameGetDbName(&name, varDataVal(db));
      varDataSetLen(db, strlen(varDataVal(db)));
H
Haojun Liao 已提交
626

627 628
      pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
      colDataAppend(pColInfo, numOfRows, (const char *)db, false);
H
Haojun Liao 已提交
629

630 631 632
      uint32_t val = 0;
      pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
      colDataAppend(pColInfo, numOfRows, (const char *)&val, false);
H
Haojun Liao 已提交
633

634
      char buf[20] = {0};
S
Shengliang Guan 已提交
635
      STR_TO_VARSTR(buf, syncStr(pVgid->role));
636 637
      pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
      colDataAppend(pColInfo, numOfRows, (const char *)buf, false);
H
Haojun Liao 已提交
638

639 640
      pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
      colDataAppend(pColInfo, numOfRows, (const char *)&pVgroup->replica, false);  // onlines
H
Haojun Liao 已提交
641

S
Shengliang Guan 已提交
642 643 644 645 646 647
      numOfRows++;
    }

    sdbRelease(pSdb, pVgroup);
  }

648
  pShow->numOfRows += numOfRows;
S
Shengliang Guan 已提交
649 650 651 652 653 654
  return numOfRows;
}

static void mndCancelGetNextVnode(SMnode *pMnode, void *pIter) {
  SSdb *pSdb = pMnode->pSdb;
  sdbCancelFetch(pSdb, pIter);
L
Liu Jicong 已提交
655
}