mndVgroup.c 21.7 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 37
static int32_t mndProcessCreateVnodeRsp(SNodeMsg *pRsp);
static int32_t mndProcessAlterVnodeRsp(SNodeMsg *pRsp);
static int32_t mndProcessDropVnodeRsp(SNodeMsg *pRsp);
static int32_t mndProcessSyncVnodeRsp(SNodeMsg *pRsp);
static int32_t mndProcessCompactVnodeRsp(SNodeMsg *pRsp);

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

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

H
Hongze Cheng 已提交
52 53 54 55 56
  mndSetMsgHandle(pMnode, TDMT_DND_CREATE_VNODE_RSP, mndProcessCreateVnodeRsp);
  mndSetMsgHandle(pMnode, TDMT_DND_ALTER_VNODE_RSP, mndProcessAlterVnodeRsp);
  mndSetMsgHandle(pMnode, TDMT_DND_DROP_VNODE_RSP, mndProcessDropVnodeRsp);
  mndSetMsgHandle(pMnode, TDMT_DND_SYNC_VNODE_RSP, mndProcessSyncVnodeRsp);
  mndSetMsgHandle(pMnode, TDMT_DND_COMPACT_VNODE_RSP, mndProcessCompactVnodeRsp);
S
Shengliang Guan 已提交
57 58 59

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

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

void mndCleanupVgroup(SMnode *pMnode) {}

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

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

  int32_t dataPos = 0;
75 76 77 78 79 80 81 82 83
  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 已提交
84 85
  for (int8_t i = 0; i < pVgroup->replica; ++i) {
    SVnodeGid *pVgid = &pVgroup->vnodeGid[i];
86 87 88 89 90 91 92 93 94 95 96 97
    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 已提交
98 99
  }

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

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

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

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

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

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

  int32_t dataPos = 0;
122 123 124 125 126 127 128 129 130
  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 已提交
131 132
  for (int8_t i = 0; i < pVgroup->replica; ++i) {
    SVnodeGid *pVgid = &pVgroup->vnodeGid[i];
133
    SDB_GET_INT32(pRaw, dataPos, &pVgid->dnodeId, VG_DECODE_OVER)
S
Shengliang Guan 已提交
134 135 136
    if (pVgroup->replica == 1) {
      pVgid->role = TAOS_SYNC_STATE_LEADER;
    }
S
Shengliang Guan 已提交
137
  }
138 139 140 141 142 143 144
  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 已提交
145
    taosMemoryFreeClear(pRow);
146 147
    return NULL;
  }
S
Shengliang Guan 已提交
148

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

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

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

S
Shengliang Guan 已提交
163 164 165 166 167 168 169 170
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 已提交
171 172 173 174
  return 0;
}

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

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

S
Shengliang Guan 已提交
188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212
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;
  createReq.quorum = pDb->cfg.quorum;
  createReq.update = pDb->cfg.update;
  createReq.cacheLastRow = pDb->cfg.cacheLastRow;
  createReq.replica = pVgroup->replica;
  createReq.selfIndex = -1;
L
Liu Jicong 已提交
213
  createReq.streamMode = pVgroup->streamMode;
D
dapan1121 已提交
214 215 216
  createReq.hashBegin = pVgroup->hashBegin;
  createReq.hashEnd = pVgroup->hashEnd;
  createReq.hashMethod = pDb->hashMethod;
S
sma  
Shengliang Guan 已提交
217 218
  createReq.numOfRetensions = pDb->cfg.numOfRetensions;
  createReq.pRetensions = pDb->cfg.pRetensions;
S
Shengliang Guan 已提交
219 220

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

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

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

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

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

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

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

L
Liu Jicong 已提交
260
void *mndBuildDropVnodeReq(SMnode *pMnode, SDnodeObj *pDnode, SDbObj *pDb, SVgObj *pVgroup, int32_t *pContLen) {
S
Shengliang Guan 已提交
261 262 263 264 265 266 267 268
  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 已提交
269 270 271 272
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return NULL;
  }

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

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

S
Shengliang Guan 已提交
284 285 286 287 288 289 290 291 292 293 294 295
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 已提交
296 297
  bool    isMnode = mndIsMnode(pMnode, pDnode->id);
  pDnode->numOfVnodes = mndGetVnodesNum(pMnode, pDnode->id);
S
Shengliang Guan 已提交
298 299 300 301 302 303 304 305

  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 已提交
306 307 308
  if (online && pDnode->numOfSupportVnodes > 0) {
    taosArrayPush(pArray, pDnode);
  }
S
Shengliang Guan 已提交
309 310 311
  return true;
}

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

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

S
Shengliang Guan 已提交
322 323
  sdbTraverse(pSdb, SDB_DNODE, mndResetDnodesArrayFp, NULL, NULL, NULL);
  sdbTraverse(pSdb, SDB_DNODE, mndBuildDnodesArrayFp, pArray, NULL, NULL);
S
Shengliang Guan 已提交
324 325 326 327 328 329
  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 已提交
330
  return d1Score >= d2Score ? 1 : 0;
S
Shengliang Guan 已提交
331 332 333 334 335 336 337 338 339
}

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 已提交
340 341 342 343 344 345 346 347
  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 已提交
348 349 350 351 352 353 354 355 356 357 358 359 360 361 362
  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 已提交
363
    mDebug("db:%s, vgId:%d, vn:%d dnode:%d is alloced", pVgroup->dbName, pVgroup->vgId, v, pVgid->dnodeId);
S
Shengliang Guan 已提交
364
    pDnode->numOfVnodes++;
S
Shengliang Guan 已提交
365
  }
S
Shengliang Guan 已提交
366

S
Shengliang Guan 已提交
367 368
  return 0;
}
369

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

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

S
Shengliang Guan 已提交
381 382 383 384 385 386 387 388
  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 已提交
389
  int32_t  allocedVgroups = 0;
S
Shengliang Guan 已提交
390 391 392
  int32_t  maxVgId = sdbGetMaxId(pMnode->pSdb, SDB_VGROUP);
  uint32_t hashMin = 0;
  uint32_t hashMax = UINT32_MAX;
393
  uint32_t hashInterval = (hashMax - hashMin) / pDb->cfg.numOfVgroups;
S
Shengliang Guan 已提交
394

395 396
  if (maxVgId < 2) maxVgId = 2;

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

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

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

S
Shengliang Guan 已提交
420
    allocedVgroups++;
421 422
  }

S
Shengliang Guan 已提交
423
  *ppVgroups = pVgroups;
S
Shengliang Guan 已提交
424 425 426 427 428
  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 已提交
429
  if (code != 0) taosMemoryFree(pVgroups);
S
Shengliang Guan 已提交
430 431
  taosArrayDestroy(pArray);
  return code;
432 433
}

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

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

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

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

  return epset;
}

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

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

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

S
Shengliang Guan 已提交
468
static int32_t mndProcessSyncVnodeRsp(SNodeMsg *pRsp) { return 0; }
S
Shengliang Guan 已提交
469

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

S
Shengliang Guan 已提交
472 473 474 475 476 477 478
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 已提交
479
    *pReplica = TMAX(*pReplica, pVgroup->replica);
S
Shengliang Guan 已提交
480 481 482 483 484 485
    (*pNumOfVgroups)++;
  }

  return true;
}

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

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

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

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

S
Shengliang Guan 已提交
516 517 518 519
  while (numOfRows < rows) {
    pShow->pIter = sdbFetch(pSdb, SDB_VGROUP, pShow->pIter, (void **)&pVgroup);
    if (pShow->pIter == NULL) break;

H
Haojun Liao 已提交
520 521 522 523 524
    if (pDb != NULL && pVgroup->dbUid != pDb->uid) {
      continue;
    }

    cols = 0;
525 526
    SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    colDataAppend(pColInfo, numOfRows, (const char *)&pVgroup->vgId, false);
H
Haojun Liao 已提交
527 528

    SName name = {0};
529 530 531 532
    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 已提交
533

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

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

540 541 542 543 544
    // status
    char buf[10] = {0};
    STR_TO_VARSTR(buf, "ready");  // TODO
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    colDataAppend(pColInfo, numOfRows, buf, false);
S
Shengliang Guan 已提交
545

546
    // onlines
547
    int32_t onlines = pVgroup->replica;
548
    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
549
    colDataAppend(pColInfo, numOfRows, (const char *)&onlines, false);
H
Haojun Liao 已提交
550

551 552
    // default 3 replica
    for (int32_t i = 0; i < 3; ++i) {
553 554

      pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
555 556 557 558 559 560 561 562 563 564 565 566 567 568
      if (i < pVgroup->replica) {
        colDataAppend(pColInfo, numOfRows, (const char *)&pVgroup->vnodeGid[i].dnodeId, false);

        char        buf1[20] = {0};
        const char *role = mndGetRoleStr(pVgroup->vnodeGid[i].role);
        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 已提交
569
    }
S
Shengliang Guan 已提交
570

571 572 573 574 575 576 577 578 579
    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 已提交
580
    numOfRows++;
S
Shengliang Guan 已提交
581 582 583
    sdbRelease(pSdb, pVgroup);
  }

H
Haojun Liao 已提交
584 585 586 587
  if (pDb != NULL) {
    mndReleaseDb(pMnode, pDb);
  }

588
  pShow->numOfRows += numOfRows;
S
Shengliang Guan 已提交
589 590 591 592 593 594 595 596
  return numOfRows;
}

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

S
Shengliang Guan 已提交
597 598 599 600
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 已提交
601

S
Shengliang Guan 已提交
602 603 604
  for (int32_t v = 0; v < pVgroup->replica; ++v) {
    if (pVgroup->vnodeGid[v].dnodeId == dnodeId) {
      (*pNumOfVnodes)++;
S
Shengliang Guan 已提交
605
    }
S
Shengliang Guan 已提交
606 607
  }

S
Shengliang Guan 已提交
608 609 610 611 612 613
  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 已提交
614
  return numOfVnodes;
S
Shengliang Guan 已提交
615 616
}

617
static int32_t mndRetrieveVnodes(SNodeMsg *pReq, SShowObj *pShow, SSDataBlock* pBlock, int32_t rows) {
S
Shengliang Guan 已提交
618
  SMnode *pMnode = pReq->pNode;
S
Shengliang Guan 已提交
619 620 621 622
  SSdb   *pSdb = pMnode->pSdb;
  int32_t numOfRows = 0;
  SVgObj *pVgroup = NULL;
  int32_t cols = 0;
H
Haojun Liao 已提交
623
//  int32_t dnodeId = pShow->replica;
S
Shengliang Guan 已提交
624 625 626 627 628

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

S
Shengliang Guan 已提交
629
    for (int32_t i = 0; i < pVgroup->replica && numOfRows < rows; ++i) {
S
Shengliang Guan 已提交
630 631 632
      SVnodeGid *pVgid = &pVgroup->vnodeGid[i];
      cols = 0;

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

H
Haojun Liao 已提交
636
      SName name = {0};
637 638 639 640
      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 已提交
641

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

645 646 647
      uint32_t val = 0;
      pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
      colDataAppend(pColInfo, numOfRows, (const char *)&val, false);
H
Haojun Liao 已提交
648

649 650 651 652
      char buf[20] = {0};
      STR_TO_VARSTR(buf, mndGetRoleStr(pVgid->role));
      pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
      colDataAppend(pColInfo, numOfRows, (const char *)buf, false);
H
Haojun Liao 已提交
653

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

S
Shengliang Guan 已提交
657 658 659 660 661 662
      numOfRows++;
    }

    sdbRelease(pSdb, pVgroup);
  }

663
  pShow->numOfRows += numOfRows;
S
Shengliang Guan 已提交
664 665 666 667 668 669
  return numOfRows;
}

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