vmHandle.c 24.1 KB
Newer Older
S
shm  
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
shm  
Shengliang Guan 已提交
17
#include "vmInt.h"
S
shm  
Shengliang Guan 已提交
18

C
Cary Xu 已提交
19
void vmGetVnodeLoads(SVnodeMgmt *pMgmt, SMonVloadInfo *pInfo, bool isReset) {
S
Shengliang Guan 已提交
20 21 22
  pInfo->pVloads = taosArrayInit(pMgmt->state.totalVnodes, sizeof(SVnodeLoad));
  if (pInfo->pVloads == NULL) return;

23
  tfsUpdateSize(pMgmt->pTfs);
X
Xiaoyu Wang 已提交
24

25
  taosThreadRwlockRdlock(&pMgmt->lock);
S
Shengliang Guan 已提交
26 27 28 29 30 31 32 33 34

  void *pIter = taosHashIterate(pMgmt->hash, NULL);
  while (pIter) {
    SVnodeObj **ppVnode = pIter;
    if (ppVnode == NULL || *ppVnode == NULL) continue;

    SVnodeObj *pVnode = *ppVnode;
    SVnodeLoad vload = {0};
    vnodeGetLoad(pVnode->pImpl, &vload);
C
Cary Xu 已提交
35
    if (isReset) vnodeResetLoad(pVnode->pImpl, &vload);
S
Shengliang Guan 已提交
36 37 38 39
    taosArrayPush(pInfo->pVloads, &vload);
    pIter = taosHashIterate(pMgmt->hash, pIter);
  }

40
  taosThreadRwlockUnlock(&pMgmt->lock);
S
Shengliang Guan 已提交
41 42
}

S
Shengliang Guan 已提交
43
void vmGetMonitorInfo(SVnodeMgmt *pMgmt, SMonVmInfo *pInfo) {
S
Shengliang Guan 已提交
44
  SMonVloadInfo vloads = {0};
C
Cary Xu 已提交
45
  vmGetVnodeLoads(pMgmt, &vloads, true);
S
Shengliang Guan 已提交
46 47 48

  SArray *pVloads = vloads.pVloads;
  if (pVloads == NULL) return;
S
Shengliang Guan 已提交
49 50 51 52 53 54 55 56 57

  int32_t totalVnodes = 0;
  int32_t masterNum = 0;
  int64_t numOfSelectReqs = 0;
  int64_t numOfInsertReqs = 0;
  int64_t numOfInsertSuccessReqs = 0;
  int64_t numOfBatchInsertReqs = 0;
  int64_t numOfBatchInsertSuccessReqs = 0;

S
Shengliang Guan 已提交
58 59
  for (int32_t i = 0; i < taosArrayGetSize(pVloads); ++i) {
    SVnodeLoad *pLoad = taosArrayGet(pVloads, i);
S
Shengliang Guan 已提交
60 61 62 63 64 65 66 67 68 69 70
    numOfSelectReqs += pLoad->numOfSelectReqs;
    numOfInsertReqs += pLoad->numOfInsertReqs;
    numOfInsertSuccessReqs += pLoad->numOfInsertSuccessReqs;
    numOfBatchInsertReqs += pLoad->numOfBatchInsertReqs;
    numOfBatchInsertSuccessReqs += pLoad->numOfBatchInsertSuccessReqs;
    if (pLoad->syncState == TAOS_SYNC_STATE_LEADER) masterNum++;
    totalVnodes++;
  }

  pInfo->vstat.totalVnodes = totalVnodes;
  pInfo->vstat.masterNum = masterNum;
71
  pInfo->vstat.numOfSelectReqs = numOfSelectReqs;
C
Cary Xu 已提交
72 73 74 75
  pInfo->vstat.numOfInsertReqs = numOfInsertReqs;                          // delta
  pInfo->vstat.numOfInsertSuccessReqs = numOfInsertSuccessReqs;            // delta
  pInfo->vstat.numOfBatchInsertReqs = numOfBatchInsertReqs;                // delta
  pInfo->vstat.numOfBatchInsertSuccessReqs = numOfBatchInsertSuccessReqs;  // delta
S
Shengliang Guan 已提交
76 77 78 79 80 81 82
  pMgmt->state.totalVnodes = totalVnodes;
  pMgmt->state.masterNum = masterNum;
  pMgmt->state.numOfSelectReqs = numOfSelectReqs;
  pMgmt->state.numOfInsertReqs = numOfInsertReqs;
  pMgmt->state.numOfInsertSuccessReqs = numOfInsertSuccessReqs;
  pMgmt->state.numOfBatchInsertReqs = numOfBatchInsertReqs;
  pMgmt->state.numOfBatchInsertSuccessReqs = numOfBatchInsertSuccessReqs;
S
Shengliang Guan 已提交
83

84
  tfsGetMonitorInfo(pMgmt->pTfs, &pInfo->tfs);
S
Shengliang Guan 已提交
85
  taosArrayDestroy(pVloads);
86 87
}

S
shm  
Shengliang Guan 已提交
88
static void vmGenerateVnodeCfg(SCreateVnodeReq *pCreate, SVnodeCfg *pCfg) {
H
Hongze Cheng 已提交
89 90
  memcpy(pCfg, &vnodeCfgDefault, sizeof(SVnodeCfg));

S
shm  
Shengliang Guan 已提交
91
  pCfg->vgId = pCreate->vgId;
S
Shengliang Guan 已提交
92 93
  tstrncpy(pCfg->dbname, pCreate->db, sizeof(pCfg->dbname));
  pCfg->dbId = pCreate->dbUid;
H
Hongze Cheng 已提交
94 95
  pCfg->szPage = pCreate->pageSize * 1024;
  pCfg->szCache = pCreate->pages;
96 97
  pCfg->cacheLast = pCreate->cacheLast;
  pCfg->cacheLastSize = pCreate->cacheLastSize;
H
Hongze Cheng 已提交
98
  pCfg->szBuf = (uint64_t)pCreate->buffer * 1024 * 1024;
S
shm  
Shengliang Guan 已提交
99
  pCfg->isWeak = true;
C
Cary Xu 已提交
100
  pCfg->isTsma = pCreate->isTsma;
C
Cary Xu 已提交
101
  pCfg->tsdbCfg.compression = pCreate->compression;
H
Hongze Cheng 已提交
102
  pCfg->tsdbCfg.precision = pCreate->precision;
103 104 105 106
  pCfg->tsdbCfg.days = pCreate->daysPerFile;
  pCfg->tsdbCfg.keep0 = pCreate->daysToKeep0;
  pCfg->tsdbCfg.keep1 = pCreate->daysToKeep1;
  pCfg->tsdbCfg.keep2 = pCreate->daysToKeep2;
H
Hongze Cheng 已提交
107 108
  pCfg->tsdbCfg.minRows = pCreate->minRows;
  pCfg->tsdbCfg.maxRows = pCreate->maxRows;
C
Cary Xu 已提交
109
  for (size_t i = 0; i < taosArrayGetSize(pCreate->pRetensions); ++i) {
C
Cary Xu 已提交
110 111 112 113 114
    SRetention *pRetention = &pCfg->tsdbCfg.retentions[i];
    memcpy(pRetention, taosArrayGet(pCreate->pRetensions, i), sizeof(SRetention));
    if (i == 0) {
      if ((pRetention->freq > 0 && pRetention->keep > 0)) pCfg->isRsma = 1;
    }
C
Cary Xu 已提交
115
  }
C
Cary Xu 已提交
116

S
shm  
Shengliang Guan 已提交
117
  pCfg->walCfg.vgId = pCreate->vgId;
118
  pCfg->walCfg.fsyncPeriod = pCreate->walFsyncPeriod;
S
Shengliang Guan 已提交
119
  pCfg->walCfg.retentionPeriod = pCreate->walRetentionPeriod;
L
Liu Jicong 已提交
120 121
  pCfg->walCfg.rollPeriod = pCreate->walRollPeriod;
  pCfg->walCfg.retentionSize = pCreate->walRetentionSize;
S
Shengliang Guan 已提交
122 123 124
  pCfg->walCfg.segSize = pCreate->walSegmentSize;
  pCfg->walCfg.level = pCreate->walLevel;

H
Hongze Cheng 已提交
125
  pCfg->sttTrigger = pCreate->sstTrigger;
S
shm  
Shengliang Guan 已提交
126 127 128
  pCfg->hashBegin = pCreate->hashBegin;
  pCfg->hashEnd = pCreate->hashEnd;
  pCfg->hashMethod = pCreate->hashMethod;
129 130
  pCfg->hashPrefix = pCreate->hashPrefix;
  pCfg->hashSuffix = pCreate->hashSuffix;
131
  pCfg->tsdbPageSize = pCreate->tsdbPageSize * 1024;
M
Minghao Li 已提交
132

S
Shengliang Guan 已提交
133
  pCfg->standby = 0;
M
Minghao Li 已提交
134 135
  pCfg->syncCfg.myIndex = pCreate->selfIndex;
  pCfg->syncCfg.replicaNum = pCreate->replica;
S
Shengliang Guan 已提交
136
  memset(&pCfg->syncCfg.nodeInfo, 0, sizeof(pCfg->syncCfg.nodeInfo));
137
  for (int32_t i = 0; i < pCreate->replica; ++i) {
138
    SNodeInfo *pNode = &pCfg->syncCfg.nodeInfo[i];
139
    pNode->nodeId = pCreate->replicas[i].id;
140
    pNode->nodePort = pCreate->replicas[i].port;
141
    tstrncpy(pNode->nodeFqdn, pCreate->replicas[i].fqdn, TSDB_FQDN_LEN);
142
    tmsgUpdateDnodeInfo(&pNode->nodeId, &pNode->clusterId, pNode->nodeFqdn, &pNode->nodePort);
M
Minghao Li 已提交
143
  }
S
shm  
Shengliang Guan 已提交
144
}
S
shm  
Shengliang Guan 已提交
145

S
Shengliang 已提交
146
static void vmGenerateWrapperCfg(SVnodeMgmt *pMgmt, SCreateVnodeReq *pCreate, SWrapperCfg *pCfg) {
S
shm  
Shengliang Guan 已提交
147 148
  pCfg->vgId = pCreate->vgId;
  pCfg->vgVersion = pCreate->vgVersion;
S
Shengliang Guan 已提交
149 150
  pCfg->dropped = 0;
  snprintf(pCfg->path, sizeof(pCfg->path), "%s%svnode%d", pMgmt->path, TD_DIRSEP, pCreate->vgId);
S
shm  
Shengliang Guan 已提交
151 152
}

C
Cary Xu 已提交
153 154 155 156 157 158 159 160 161
static int32_t vmTsmaAdjustDays(SVnodeCfg *pCfg, SCreateVnodeReq *pReq) {
  if (pReq->isTsma) {
    SMsgHead *smaMsg = pReq->pTsma;
    uint32_t  contLen = (uint32_t)(htonl(smaMsg->contLen) - sizeof(SMsgHead));
    return smaGetTSmaDays(pCfg, POINTER_SHIFT(smaMsg, sizeof(SMsgHead)), contLen, &pCfg->tsdbCfg.days);
  }
  return 0;
}

K
kailixu 已提交
162
#if 0
C
Cary Xu 已提交
163 164 165 166 167 168 169 170
static int32_t vmTsmaProcessCreate(SVnode *pVnode, SCreateVnodeReq *pReq) {
  if (pReq->isTsma) {
    SMsgHead *smaMsg = pReq->pTsma;
    uint32_t  contLen = (uint32_t)(htonl(smaMsg->contLen) - sizeof(SMsgHead));
    return vnodeProcessCreateTSma(pVnode, POINTER_SHIFT(smaMsg, sizeof(SMsgHead)), contLen);
  }
  return 0;
}
K
kailixu 已提交
171
#endif
C
Cary Xu 已提交
172

S
Shengliang Guan 已提交
173
int32_t vmProcessCreateVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
174
  SCreateVnodeReq req = {0};
175 176
  SVnodeCfg       vnodeCfg = {0};
  SWrapperCfg     wrapperCfg = {0};
S
Shengliang Guan 已提交
177 178
  int32_t         code = -1;
  char            path[TSDB_FILENAME_LEN] = {0};
H
Hongze Cheng 已提交
179

180
  if (tDeserializeSCreateVnodeReq(pMsg->pCont, pMsg->contLen, &req) != 0) {
S
shm  
Shengliang Guan 已提交
181 182 183 184
    terrno = TSDB_CODE_INVALID_MSG;
    return -1;
  }

S
Shengliang Guan 已提交
185 186 187 188 189 190 191 192 193 194 195
  dInfo("vgId:%d, start to create vnode, page:%d pageSize:%d buffer:%d szPage:%d szBuf:%" PRIu64
        ", cacheLast:%d cacheLastSize:%d sstTrigger:%d tsdbPageSize:%d %d dbname:%s dbId:%" PRId64
        ", days:%d keep0:%d keep1:%d keep2:%d tsma:%d precision:%d compression:%d minRows:%d maxRows:%d"
        ", wal fsync:%d level:%d retentionPeriod:%d retentionSize:%" PRId64 " rollPeriod:%d segSize:%" PRId64
        ", hash method:%d begin:%u end:%u prefix:%d surfix:%d replica:%d selfIndex:%d strict:%d",
        req.vgId, req.pages, req.pageSize, req.buffer, req.pageSize * 1024, (uint64_t)req.buffer * 1024 * 1024,
        req.cacheLast, req.cacheLastSize, req.sstTrigger, req.tsdbPageSize, req.tsdbPageSize * 1024, req.db, req.dbUid,
        req.daysPerFile, req.daysToKeep0, req.daysToKeep1, req.daysToKeep2, req.isTsma, req.precision, req.compression,
        req.minRows, req.maxRows, req.walFsyncPeriod, req.walLevel, req.walRetentionPeriod, req.walRetentionSize,
        req.walRollPeriod, req.walSegmentSize, req.hashMethod, req.hashBegin, req.hashEnd, req.hashPrefix,
        req.hashSuffix, req.replica, req.selfIndex, req.strict);
196
  for (int32_t i = 0; i < req.replica; ++i) {
197 198
    dInfo("vgId:%d, replica:%d ep:%s:%u dnode:%d", req.vgId, i, req.replicas[i].fqdn, req.replicas[i].port,
          req.replicas[i].id);
199
  }
200 201 202 203 204 205 206 207 208 209

  SReplica *pReplica = &req.replicas[req.selfIndex];
  if (pReplica->id != pMgmt->pData->dnodeId || pReplica->port != tsServerPort ||
      strcmp(pReplica->fqdn, tsLocalFqdn) != 0) {
    terrno = TSDB_CODE_INVALID_MSG;
    dError("vgId:%d, dnodeId:%d ep:%s:%u not matched with local dnode", req.vgId, pReplica->id, pReplica->fqdn,
           pReplica->port);
    return -1;
  }

210 211 212 213
  vmGenerateVnodeCfg(&req, &vnodeCfg);

  if (vmTsmaAdjustDays(&vnodeCfg, &req) < 0) {
    dError("vgId:%d, failed to adjust tsma days since %s", req.vgId, terrstr());
C
Cary Xu 已提交
214 215
    code = terrno;
    goto _OVER;
C
Cary Xu 已提交
216 217
  }

218
  vmGenerateWrapperCfg(pMgmt, &req, &wrapperCfg);
S
shm  
Shengliang Guan 已提交
219

220
  SVnodeObj *pVnode = vmAcquireVnode(pMgmt, req.vgId);
S
shm  
Shengliang Guan 已提交
221
  if (pVnode != NULL) {
222
    dInfo("vgId:%d, already exist", req.vgId);
223
    tFreeSCreateVnodeReq(&req);
S
shm  
Shengliang Guan 已提交
224
    vmReleaseVnode(pMgmt, pVnode);
225
    terrno = TSDB_CODE_VND_ALREADY_EXIST;
C
Cary Xu 已提交
226
    code = terrno;
227
    return 0;
S
shm  
Shengliang Guan 已提交
228 229
  }

H
Hongze Cheng 已提交
230 231
  snprintf(path, TSDB_FILENAME_LEN, "vnode%svnode%d", TD_DIRSEP, vnodeCfg.vgId);
  if (vnodeCreate(path, &vnodeCfg, pMgmt->pTfs) < 0) {
232 233
    tFreeSCreateVnodeReq(&req);
    dError("vgId:%d, failed to create vnode since %s", req.vgId, terrstr());
C
Cary Xu 已提交
234 235
    code = terrno;
    goto _OVER;
H
Hongze Cheng 已提交
236 237
  }

S
Shengliang 已提交
238
  SVnode *pImpl = vnodeOpen(path, pMgmt->pTfs, pMgmt->msgCb);
S
shm  
Shengliang Guan 已提交
239
  if (pImpl == NULL) {
240
    dError("vgId:%d, failed to open vnode since %s", req.vgId, terrstr());
241
    code = terrno;
S
Shengliang Guan 已提交
242
    goto _OVER;
S
shm  
Shengliang Guan 已提交
243 244
  }

S
Shengliang Guan 已提交
245
  code = vmOpenVnode(pMgmt, &wrapperCfg, pImpl);
S
shm  
Shengliang Guan 已提交
246
  if (code != 0) {
247
    dError("vgId:%d, failed to open vnode since %s", req.vgId, terrstr());
C
Cary Xu 已提交
248
    code = terrno;
S
Shengliang Guan 已提交
249
    goto _OVER;
S
shm  
Shengliang Guan 已提交
250 251
  }

K
kailixu 已提交
252
#if 0
253
  code = vmTsmaProcessCreate(pImpl, &req);
C
Cary Xu 已提交
254
  if (code != 0) {
255
    dError("vgId:%d, failed to create tsma since %s", req.vgId, terrstr());
C
Cary Xu 已提交
256 257
    code = terrno;
    goto _OVER;
C
Cary Xu 已提交
258
  }
K
kailixu 已提交
259
#endif
C
Cary Xu 已提交
260

L
Li Minghao 已提交
261 262
  code = vnodeStart(pImpl);
  if (code != 0) {
263
    dError("vgId:%d, failed to start sync since %s", req.vgId, terrstr());
S
Shengliang Guan 已提交
264
    goto _OVER;
L
Li Minghao 已提交
265 266
  }

S
Shengliang Guan 已提交
267
  code = vmWriteVnodeListToFile(pMgmt);
C
Cary Xu 已提交
268 269 270 271
  if (code != 0) {
    code = terrno;
    goto _OVER;
  }
S
Shengliang Guan 已提交
272 273

_OVER:
S
shm  
Shengliang Guan 已提交
274 275
  if (code != 0) {
    vnodeClose(pImpl);
H
refact  
Hongze Cheng 已提交
276
    vnodeDestroy(path, pMgmt->pTfs);
S
Shengliang Guan 已提交
277
  } else {
278
    dInfo("vgId:%d, vnode is created", req.vgId);
S
shm  
Shengliang Guan 已提交
279 280
  }

281
  tFreeSCreateVnodeReq(&req);
S
Shengliang Guan 已提交
282 283
  terrno = code;
  return code;
S
shm  
Shengliang Guan 已提交
284 285
}

286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306
int32_t vmProcessDisableVnodeWriteReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
  SDisableVnodeWriteReq req = {0};
  if (tDeserializeSDisableVnodeWriteReq(pMsg->pCont, pMsg->contLen, &req) != 0) {
    terrno = TSDB_CODE_INVALID_MSG;
    return -1;
  }

  dInfo("vgId:%d, vnode write disable:%d", req.vgId, req.disable);

  SVnodeObj *pVnode = vmAcquireVnode(pMgmt, req.vgId);
  if (pVnode == NULL) {
    dError("vgId:%d, failed to disable write since %s", req.vgId, terrstr());
    terrno = TSDB_CODE_VND_NOT_EXIST;
    return -1;
  }

  pVnode->disable = req.disable;
  vmReleaseVnode(pMgmt, pVnode);
  return 0;
}

307 308 309 310 311 312 313
int32_t vmProcessAlterHashRangeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
  SAlterVnodeHashRangeReq req = {0};
  if (tDeserializeSAlterVnodeHashRangeReq(pMsg->pCont, pMsg->contLen, &req) != 0) {
    terrno = TSDB_CODE_INVALID_MSG;
    return -1;
  }

314 315
  int32_t srcVgId = req.srcVgId;
  int32_t dstVgId = req.dstVgId;
316
  dInfo("vgId:%d, start to alter vnode hashrange:[%u, %u], dstVgId:%d", req.srcVgId, req.hashBegin, req.hashEnd,
317
        req.dstVgId);
318

319 320 321 322 323 324 325
  SVnodeObj *pVnode = vmAcquireVnode(pMgmt, srcVgId);
  if (pVnode == NULL) {
    dError("vgId:%d, failed to alter hashrange since %s", srcVgId, terrstr());
    terrno = TSDB_CODE_VND_NOT_EXIST;
    return -1;
  }

326 327 328 329 330 331 332 333
  SWrapperCfg wrapperCfg = {
      .dropped = pVnode->dropped,
      .vgId = dstVgId,
      .vgVersion = pVnode->vgVersion,
  };
  tstrncpy(wrapperCfg.path, pVnode->path, sizeof(wrapperCfg.path));

  dInfo("vgId:%d, close vnode", srcVgId);
S
Shengliang Guan 已提交
334
  vmCloseVnode(pMgmt, pVnode, true);
335 336 337 338 339 340

  char srcPath[TSDB_FILENAME_LEN] = {0};
  char dstPath[TSDB_FILENAME_LEN] = {0};
  snprintf(srcPath, TSDB_FILENAME_LEN, "vnode%svnode%d", TD_DIRSEP, srcVgId);
  snprintf(dstPath, TSDB_FILENAME_LEN, "vnode%svnode%d", TD_DIRSEP, dstVgId);

341
  dInfo("vgId:%d, alter vnode hashrange at %s", srcVgId, srcPath);
342 343 344 345 346
  if (vnodeAlterHashRange(srcPath, dstPath, &req, pMgmt->pTfs) < 0) {
    dError("vgId:%d, failed to alter vnode hashrange since %s", srcVgId, terrstr());
    return -1;
  }

347
  dInfo("vgId:%d, open vnode", dstVgId);
348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363
  SVnode *pImpl = vnodeOpen(dstPath, pMgmt->pTfs, pMgmt->msgCb);
  if (pImpl == NULL) {
    dError("vgId:%d, failed to open vnode at %s since %s", dstVgId, dstPath, terrstr());
    return -1;
  }

  if (vmOpenVnode(pMgmt, &wrapperCfg, pImpl) != 0) {
    dError("vgId:%d, failed to open vnode mgmt since %s", dstVgId, terrstr());
    return -1;
  }

  if (vnodeStart(pImpl) != 0) {
    dError("vgId:%d, failed to start sync since %s", dstVgId, terrstr());
    return -1;
  }

S
Shengliang Guan 已提交
364 365 366 367 368
  if (vmWriteVnodeListToFile(pMgmt) != 0) {
    dError("vgId:%d, failed to write vnode list since %s", dstVgId, terrstr());
    return -1;
  }

369
  dInfo("vgId:%d, vnode hashrange is altered", dstVgId);
370 371 372 373
  return 0;
}

int32_t vmProcessAlterVnodeReplicaReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
S
Shengliang Guan 已提交
374
  SAlterVnodeReplicaReq alterReq = {0};
S
Shengliang Guan 已提交
375
  if (tDeserializeSAlterVnodeReplicaReq(pMsg->pCont, pMsg->contLen, &alterReq) != 0) {
S
Shengliang Guan 已提交
376 377 378 379 380
    terrno = TSDB_CODE_INVALID_MSG;
    return -1;
  }

  int32_t vgId = alterReq.vgId;
381
  dInfo("vgId:%d, start to alter vnode replica:%d selfIndex:%d strict:%d", vgId, alterReq.replica, alterReq.selfIndex,
382
        alterReq.strict);
383
  for (int32_t i = 0; i < alterReq.replica; ++i) {
384
    SReplica *pReplica = &alterReq.replicas[i];
385
    dInfo("vgId:%d, replica:%d ep:%s:%u dnode:%d", vgId, i, pReplica->fqdn, pReplica->port, pReplica->port);
386 387 388 389
  }

  if (alterReq.replica <= 0 || alterReq.selfIndex < 0 || alterReq.selfIndex >= alterReq.replica) {
    terrno = TSDB_CODE_INVALID_MSG;
390
    dError("vgId:%d, failed to alter replica since invalid msg", vgId);
391 392
    return -1;
  }
S
Shengliang Guan 已提交
393

394 395 396 397
  SReplica *pReplica = &alterReq.replicas[alterReq.selfIndex];
  if (pReplica->id != pMgmt->pData->dnodeId || pReplica->port != tsServerPort ||
      strcmp(pReplica->fqdn, tsLocalFqdn) != 0) {
    terrno = TSDB_CODE_INVALID_MSG;
398
    dError("vgId:%d, dnodeId:%d ep:%s:%u not matched with local dnode", vgId, pReplica->id, pReplica->fqdn,
399 400 401 402
           pReplica->port);
    return -1;
  }

S
Shengliang Guan 已提交
403 404 405
  SVnodeObj *pVnode = vmAcquireVnode(pMgmt, vgId);
  if (pVnode == NULL) {
    dError("vgId:%d, failed to alter replica since %s", vgId, terrstr());
406
    terrno = TSDB_CODE_VND_NOT_EXIST;
S
Shengliang Guan 已提交
407 408 409 410
    return -1;
  }

  dInfo("vgId:%d, start to close vnode", vgId);
S
Shengliang Guan 已提交
411 412 413 414 415 416
  SWrapperCfg wrapperCfg = {
      .dropped = pVnode->dropped,
      .vgId = pVnode->vgId,
      .vgVersion = pVnode->vgVersion,
  };
  tstrncpy(wrapperCfg.path, pVnode->path, sizeof(wrapperCfg.path));
S
Shengliang Guan 已提交
417
  vmCloseVnode(pMgmt, pVnode, false);
S
Shengliang Guan 已提交
418 419 420 421 422

  char path[TSDB_FILENAME_LEN] = {0};
  snprintf(path, TSDB_FILENAME_LEN, "vnode%svnode%d", TD_DIRSEP, vgId);

  dInfo("vgId:%d, start to alter vnode replica at %s", vgId, path);
423
  if (vnodeAlterReplica(path, &alterReq, pMgmt->pTfs) < 0) {
S
Shengliang Guan 已提交
424 425 426 427
    dError("vgId:%d, failed to alter vnode at %s since %s", vgId, path, terrstr());
    return -1;
  }

428
  dInfo("vgId:%d, close vnode", vgId);
S
Shengliang Guan 已提交
429 430
  SVnode *pImpl = vnodeOpen(path, pMgmt->pTfs, pMgmt->msgCb);
  if (pImpl == NULL) {
S
Shengliang Guan 已提交
431 432 433 434
    dError("vgId:%d, failed to open vnode at %s since %s", vgId, path, terrstr());
    return -1;
  }

S
Shengliang Guan 已提交
435 436 437 438 439 440 441 442 443 444
  if (vmOpenVnode(pMgmt, &wrapperCfg, pImpl) != 0) {
    dError("vgId:%d, failed to open vnode mgmt since %s", vgId, terrstr());
    return -1;
  }

  if (vnodeStart(pImpl) != 0) {
    dError("vgId:%d, failed to start sync since %s", vgId, terrstr());
    return -1;
  }

S
Shengliang Guan 已提交
445 446 447 448
  dInfo("vgId:%d, vnode config is altered", vgId);
  return 0;
}

S
Shengliang Guan 已提交
449
int32_t vmProcessDropVnodeReq(SVnodeMgmt *pMgmt, SRpcMsg *pMsg) {
S
shm  
Shengliang Guan 已提交
450
  SDropVnodeReq dropReq = {0};
S
Shengliang Guan 已提交
451
  if (tDeserializeSDropVnodeReq(pMsg->pCont, pMsg->contLen, &dropReq) != 0) {
S
shm  
Shengliang Guan 已提交
452 453 454 455 456
    terrno = TSDB_CODE_INVALID_MSG;
    return -1;
  }

  int32_t vgId = dropReq.vgId;
457
  dInfo("vgId:%d, start to drop vnode", vgId);
S
shm  
Shengliang Guan 已提交
458

459 460 461 462 463 464
  if (dropReq.dnodeId != pMgmt->pData->dnodeId) {
    terrno = TSDB_CODE_INVALID_MSG;
    dError("vgId:%d, dnodeId:%d not matched with local dnode", dropReq.vgId, dropReq.dnodeId);
    return -1;
  }

S
shm  
Shengliang Guan 已提交
465 466
  SVnodeObj *pVnode = vmAcquireVnode(pMgmt, vgId);
  if (pVnode == NULL) {
467
    dInfo("vgId:%d, failed to drop since %s", vgId, terrstr());
468
    terrno = TSDB_CODE_VND_NOT_EXIST;
S
shm  
Shengliang Guan 已提交
469 470 471 472
    return -1;
  }

  pVnode->dropped = 1;
S
Shengliang Guan 已提交
473
  if (vmWriteVnodeListToFile(pMgmt) != 0) {
S
shm  
Shengliang Guan 已提交
474 475 476 477 478
    pVnode->dropped = 0;
    vmReleaseVnode(pMgmt, pVnode);
    return -1;
  }

S
Shengliang Guan 已提交
479
  vmCloseVnode(pMgmt, pVnode, false);
S
Shengliang Guan 已提交
480
  vmWriteVnodeListToFile(pMgmt);
S
shm  
Shengliang Guan 已提交
481

482
  dInfo("vgId:%d, is dropped", vgId);
S
shm  
Shengliang Guan 已提交
483 484 485
  return 0;
}

S
Shengliang Guan 已提交
486
SArray *vmGetMsgHandles() {
S
Shengliang 已提交
487
  int32_t code = -1;
S
Shengliang Guan 已提交
488
  SArray *pArray = taosArrayInit(32, sizeof(SMgmtHandle));
S
Shengliang 已提交
489 490
  if (pArray == NULL) goto _OVER;

S
Shengliang Guan 已提交
491
  if (dmSetMgmtHandle(pArray, TDMT_VND_SUBMIT, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
D
dapan1121 已提交
492
  if (dmSetMgmtHandle(pArray, TDMT_SCH_QUERY, vmPutMsgToQueryQueue, 0) == NULL) goto _OVER;
D
dapan1121 已提交
493
  if (dmSetMgmtHandle(pArray, TDMT_SCH_MERGE_QUERY, vmPutMsgToQueryQueue, 0) == NULL) goto _OVER;
D
dapan1121 已提交
494
  if (dmSetMgmtHandle(pArray, TDMT_SCH_QUERY_CONTINUE, vmPutMsgToQueryQueue, 0) == NULL) goto _OVER;
C
Cary Xu 已提交
495
  if (dmSetMgmtHandle(pArray, TDMT_VND_FETCH_RSMA, vmPutMsgToQueryQueue, 0) == NULL) goto _OVER;
C
Cary Xu 已提交
496
  if (dmSetMgmtHandle(pArray, TDMT_VND_EXEC_RSMA, vmPutMsgToQueryQueue, 0) == NULL) goto _OVER;
D
dapan1121 已提交
497
  if (dmSetMgmtHandle(pArray, TDMT_SCH_FETCH, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
D
dapan1121 已提交
498
  if (dmSetMgmtHandle(pArray, TDMT_SCH_MERGE_FETCH, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
S
Shengliang Guan 已提交
499 500 501
  if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_TABLE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_VND_UPDATE_TAG_VAL, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_VND_TABLE_META, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
D
dapan1121 已提交
502
  if (dmSetMgmtHandle(pArray, TDMT_VND_TABLE_CFG, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
D
dapan1121 已提交
503
  if (dmSetMgmtHandle(pArray, TDMT_VND_BATCH_META, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
S
Shengliang Guan 已提交
504
  if (dmSetMgmtHandle(pArray, TDMT_VND_TABLES_META, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
D
dapan1121 已提交
505 506
  if (dmSetMgmtHandle(pArray, TDMT_SCH_CANCEL_TASK, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_SCH_DROP_TASK, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
S
Shengliang Guan 已提交
507
  if (dmSetMgmtHandle(pArray, TDMT_VND_CREATE_STB, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
wmmhello's avatar
wmmhello 已提交
508
  if (dmSetMgmtHandle(pArray, TDMT_VND_DROP_TTL_TABLE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
S
Shengliang Guan 已提交
509 510 511 512 513 514 515 516
  if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_STB, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_VND_DROP_STB, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_VND_CREATE_TABLE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_VND_DROP_TABLE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_VND_CREATE_SMA, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_VND_CANCEL_SMA, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_VND_DROP_SMA, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_VND_SUBMIT_RSMA, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
L
Liu Jicong 已提交
517 518 519 520 521 522
  if (dmSetMgmtHandle(pArray, TDMT_VND_TMQ_SUBSCRIBE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_VND_TMQ_DELETE_SUB, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_VND_TMQ_COMMIT_OFFSET, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_VND_TMQ_ADD_CHECKINFO, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_VND_TMQ_DEL_CHECKINFO, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_VND_TMQ_CONSUME, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
523
  if (dmSetMgmtHandle(pArray, TDMT_VND_DELETE, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
524
  if (dmSetMgmtHandle(pArray, TDMT_VND_BATCH_DEL, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
H
Hongze Cheng 已提交
525
  if (dmSetMgmtHandle(pArray, TDMT_VND_COMMIT, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
D
dapan1121 已提交
526
  if (dmSetMgmtHandle(pArray, TDMT_SCH_QUERY_HEARTBEAT, vmPutMsgToFetchQueue, 0) == NULL) goto _OVER;
dengyihao's avatar
dengyihao 已提交
527 528
  if (dmSetMgmtHandle(pArray, TDMT_VND_CREATE_INDEX, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_VND_DROP_INDEX, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
529

L
Liu Jicong 已提交
530
  if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_DROP, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
S
Shengliang Guan 已提交
531
  if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_DEPLOY, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
L
Liu Jicong 已提交
532 533 534 535 536
  if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_RUN, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_DISPATCH, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_DISPATCH_RSP, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_STREAM_RETRIEVE, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_STREAM_RETRIEVE_RSP, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
L
Liu Jicong 已提交
537
  if (dmSetMgmtHandle(pArray, TDMT_STREAM_RECOVER_FINISH, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
538 539 540
  if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_CHECK, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_STREAM_TASK_CHECK_RSP, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_VND_STREAM_TRIGGER, vmPutMsgToStreamQueue, 0) == NULL) goto _OVER;
S
Shengliang Guan 已提交
541

S
Shengliang Guan 已提交
542
  if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_REPLICA, vmPutMsgToMgmtQueue, 0) == NULL) goto _OVER;
S
Shengliang Guan 已提交
543
  if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_CONFIG, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
544
  if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_CONFIRM, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
545
  if (dmSetMgmtHandle(pArray, TDMT_VND_DISABLE_WRITE, vmPutMsgToMgmtQueue, 0) == NULL) goto _OVER;
546
  if (dmSetMgmtHandle(pArray, TDMT_VND_ALTER_HASHRANGE, vmPutMsgToMgmtQueue, 0) == NULL) goto _OVER;
S
Shengliang Guan 已提交
547
  if (dmSetMgmtHandle(pArray, TDMT_VND_COMPACT, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
S
Shengliang Guan 已提交
548
  if (dmSetMgmtHandle(pArray, TDMT_VND_TRIM, vmPutMsgToWriteQueue, 0) == NULL) goto _OVER;
S
Shengliang Guan 已提交
549 550 551
  if (dmSetMgmtHandle(pArray, TDMT_DND_CREATE_VNODE, vmPutMsgToMgmtQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_DND_DROP_VNODE, vmPutMsgToMgmtQueue, 0) == NULL) goto _OVER;

552
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_TIMEOUT_ELECTION, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
553
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_CLIENT_REQUEST, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
554
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_CLIENT_REQUEST_BATCH, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
555 556 557 558
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_CLIENT_REQUEST_REPLY, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_REQUEST_VOTE, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_REQUEST_VOTE_REPLY, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_APPEND_ENTRIES, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
559
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_APPEND_ENTRIES_BATCH, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
560
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_APPEND_ENTRIES_REPLY, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
561
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_SNAPSHOT_SEND, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
562
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_PRE_SNAPSHOT, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
C
cadem 已提交
563
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_FORCE_FOLLOWER, vmPutMsgToSyncQueue, 0) == NULL) goto _OVER;
S
Shengliang 已提交
564

565
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_TIMEOUT, vmPutMsgToSyncRdQueue, 0) == NULL) goto _OVER;
X
Xiaoyu Wang 已提交
566
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_HEARTBEAT, vmPutMsgToSyncRdQueue, 0) == NULL) goto _OVER;
567 568 569
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_HEARTBEAT_REPLY, vmPutMsgToSyncRdQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_SNAPSHOT_RSP, vmPutMsgToSyncRdQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_SYNC_PRE_SNAPSHOT_REPLY, vmPutMsgToSyncRdQueue, 0) == NULL) goto _OVER;
S
Shengliang 已提交
570 571 572 573 574 575 576 577 578 579

  code = 0;

_OVER:
  if (code != 0) {
    taosArrayDestroy(pArray);
    return NULL;
  } else {
    return pArray;
  }
S
shm  
Shengliang Guan 已提交
580
}