dndMgmt.c 23.6 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
#include "dndMgmt.h"
S
Shengliang Guan 已提交
18 19 20 21
#include "dndBnode.h"
#include "dndMnode.h"
#include "dndQnode.h"
#include "dndSnode.h"
S
Shengliang Guan 已提交
22 23
#include "dndTransport.h"
#include "dndVnodes.h"
S
Shengliang Guan 已提交
24
#include "dndWorker.h"
S
monitor  
Shengliang Guan 已提交
25
#include "monitor.h"
S
Shengliang Guan 已提交
26

S
Shengliang Guan 已提交
27
static void dndProcessMgmtQueue(SDnode *pDnode, SRpcMsg *pMsg);
S
Shengliang Guan 已提交
28 29 30 31 32

static int32_t dndReadDnodes(SDnode *pDnode);
static int32_t dndWriteDnodes(SDnode *pDnode);
static void   *dnodeThreadRoutine(void *param);

S
Shengliang Guan 已提交
33 34 35 36
static int32_t dndProcessConfigDnodeReq(SDnode *pDnode, SRpcMsg *pReq);
static void    dndProcessStatusRsp(SDnode *pDnode, SRpcMsg *pRsp);
static void    dndProcessAuthRsp(SDnode *pDnode, SRpcMsg *pRsp);
static void    dndProcessGrantRsp(SDnode *pDnode, SRpcMsg *pRsp);
S
Shengliang Guan 已提交
37

S
Shengliang Guan 已提交
38
int32_t dndGetDnodeId(SDnode *pDnode) {
S
Shengliang Guan 已提交
39 40 41 42
  SDnodeMgmt *pMgmt = &pDnode->dmgmt;
  taosRLockLatch(&pMgmt->latch);
  int32_t dnodeId = pMgmt->dnodeId;
  taosRUnLockLatch(&pMgmt->latch);
S
Shengliang Guan 已提交
43
  return dnodeId;
S
Shengliang Guan 已提交
44 45
}

S
Shengliang Guan 已提交
46
int64_t dndGetClusterId(SDnode *pDnode) {
S
Shengliang Guan 已提交
47 48
  SDnodeMgmt *pMgmt = &pDnode->dmgmt;
  taosRLockLatch(&pMgmt->latch);
S
Shengliang Guan 已提交
49
  int64_t clusterId = pMgmt->clusterId;
S
Shengliang Guan 已提交
50
  taosRUnLockLatch(&pMgmt->latch);
S
Shengliang Guan 已提交
51
  return clusterId;
S
Shengliang Guan 已提交
52 53
}

S
Shengliang Guan 已提交
54
void dndGetDnodeEp(SDnode *pDnode, int32_t dnodeId, char *pEp, char *pFqdn, uint16_t *pPort) {
S
Shengliang Guan 已提交
55 56
  SDnodeMgmt *pMgmt = &pDnode->dmgmt;
  taosRLockLatch(&pMgmt->latch);
S
Shengliang Guan 已提交
57

S
Shengliang Guan 已提交
58
  SDnodeEp *pDnodeEp = taosHashGet(pMgmt->dnodeHash, &dnodeId, sizeof(int32_t));
S
Shengliang Guan 已提交
59 60
  if (pDnodeEp != NULL) {
    if (pPort != NULL) {
H
Haojun Liao 已提交
61
      *pPort = pDnodeEp->ep.port;
S
Shengliang Guan 已提交
62 63
    }
    if (pFqdn != NULL) {
H
Haojun Liao 已提交
64
      tstrncpy(pFqdn, pDnodeEp->ep.fqdn, TSDB_FQDN_LEN);
S
Shengliang Guan 已提交
65 66
    }
    if (pEp != NULL) {
H
Haojun Liao 已提交
67
      snprintf(pEp, TSDB_EP_LEN, "%s:%u", pDnodeEp->ep.fqdn, pDnodeEp->ep.port);
S
Shengliang Guan 已提交
68
    }
S
Shengliang Guan 已提交
69
  }
S
Shengliang Guan 已提交
70

S
Shengliang Guan 已提交
71
  taosRUnLockLatch(&pMgmt->latch);
S
Shengliang Guan 已提交
72
}
S
Shengliang Guan 已提交
73

S
Shengliang Guan 已提交
74
void dndGetMnodeEpSet(SDnode *pDnode, SEpSet *pEpSet) {
S
Shengliang Guan 已提交
75 76 77 78
  SDnodeMgmt *pMgmt = &pDnode->dmgmt;
  taosRLockLatch(&pMgmt->latch);
  *pEpSet = pMgmt->mnodeEpSet;
  taosRUnLockLatch(&pMgmt->latch);
S
Shengliang Guan 已提交
79 80
}

S
Shengliang Guan 已提交
81 82
void dndSendRedirectRsp(SDnode *pDnode, SRpcMsg *pReq) {
  tmsg_t msgType = pReq->msgType;
S
Shengliang Guan 已提交
83

S
Shengliang Guan 已提交
84
  SEpSet epSet = {0};
S
Shengliang Guan 已提交
85
  dndGetMnodeEpSet(pDnode, &epSet);
S
Shengliang Guan 已提交
86

S
Shengliang Guan 已提交
87
  dDebug("RPC %p, req:%s is redirected, num:%d use:%d", pReq->handle, TMSG_INFO(msgType), epSet.numOfEps, epSet.inUse);
S
Shengliang Guan 已提交
88
  for (int32_t i = 0; i < epSet.numOfEps; ++i) {
H
Haojun Liao 已提交
89 90
    dDebug("mnode index:%d %s:%u", i, epSet.eps[i].fqdn, epSet.eps[i].port);
    if (strcmp(epSet.eps[i].fqdn, pDnode->cfg.localFqdn) == 0 && epSet.eps[i].port == pDnode->cfg.serverPort) {
S
Shengliang Guan 已提交
91
      epSet.inUse = (i + 1) % epSet.numOfEps;
S
Shengliang Guan 已提交
92 93
    }

H
Haojun Liao 已提交
94
    epSet.eps[i].port = htons(epSet.eps[i].port);
S
Shengliang Guan 已提交
95 96
  }

S
Shengliang Guan 已提交
97
  rpcSendRedirectRsp(pReq->handle, &epSet);
S
Shengliang Guan 已提交
98 99
}

S
Shengliang Guan 已提交
100
static void dndUpdateMnodeEpSet(SDnode *pDnode, SEpSet *pEpSet) {
S
Shengliang Guan 已提交
101
  dInfo("mnode is changed, num:%d use:%d", pEpSet->numOfEps, pEpSet->inUse);
S
Shengliang Guan 已提交
102

S
Shengliang Guan 已提交
103 104
  SDnodeMgmt *pMgmt = &pDnode->dmgmt;
  taosWLockLatch(&pMgmt->latch);
S
Shengliang Guan 已提交
105

S
Shengliang Guan 已提交
106
  pMgmt->mnodeEpSet = *pEpSet;
S
Shengliang Guan 已提交
107
  for (int32_t i = 0; i < pEpSet->numOfEps; ++i) {
H
Haojun Liao 已提交
108
    dInfo("mnode index:%d %s:%u", i, pEpSet->eps[i].fqdn, pEpSet->eps[i].port);
S
Shengliang Guan 已提交
109 110
  }

S
Shengliang Guan 已提交
111
  taosWUnLockLatch(&pMgmt->latch);
S
Shengliang Guan 已提交
112 113
}

S
Shengliang Guan 已提交
114 115
static void dndPrintDnodes(SDnode *pDnode) {
  SDnodeMgmt *pMgmt = &pDnode->dmgmt;
S
Shengliang Guan 已提交
116

S
Shengliang Guan 已提交
117 118 119 120
  int32_t numOfEps = (int32_t)taosArrayGetSize(pMgmt->pDnodeEps);
  dDebug("print dnode ep list, num:%d", numOfEps);
  for (int32_t i = 0; i < numOfEps; i++) {
    SDnodeEp *pEp = taosArrayGet(pMgmt->pDnodeEps, i);
H
Haojun Liao 已提交
121
    dDebug("dnode:%d, fqdn:%s port:%u isMnode:%d", pEp->id, pEp->ep.fqdn, pEp->ep.port, pEp->isMnode);
S
Shengliang Guan 已提交
122 123 124
  }
}

S
Shengliang Guan 已提交
125
static void dndResetDnodes(SDnode *pDnode, SArray *pDnodeEps) {
S
Shengliang Guan 已提交
126
  SDnodeMgmt *pMgmt = &pDnode->dmgmt;
S
Shengliang Guan 已提交
127

S
Shengliang Guan 已提交
128 129 130 131
  if (pMgmt->pDnodeEps != pDnodeEps) {
    SArray *tmp = pMgmt->pDnodeEps;
    pMgmt->pDnodeEps = taosArrayDup(pDnodeEps);
    taosArrayDestroy(tmp);
S
Shengliang Guan 已提交
132 133
  }

S
Shengliang Guan 已提交
134
  pMgmt->mnodeEpSet.inUse = 0;
135
  pMgmt->mnodeEpSet.numOfEps = 0;
S
Shengliang Guan 已提交
136 137

  int32_t mIndex = 0;
S
Shengliang Guan 已提交
138 139 140 141
  int32_t numOfEps = (int32_t)taosArrayGetSize(pDnodeEps);

  for (int32_t i = 0; i < numOfEps; i++) {
    SDnodeEp *pDnodeEp = taosArrayGet(pDnodeEps, i);
S
Shengliang Guan 已提交
142
    if (!pDnodeEp->isMnode) continue;
S
Shengliang Guan 已提交
143
    if (mIndex >= TSDB_MAX_REPLICA) continue;
144
    pMgmt->mnodeEpSet.numOfEps++;
H
Haojun Liao 已提交
145 146

    pMgmt->mnodeEpSet.eps[mIndex] = pDnodeEp->ep;
S
Shengliang Guan 已提交
147
    mIndex++;
S
Shengliang Guan 已提交
148 149
  }

S
Shengliang Guan 已提交
150 151
  for (int32_t i = 0; i < numOfEps; i++) {
    SDnodeEp *pDnodeEp = taosArrayGet(pDnodeEps, i);
S
Shengliang Guan 已提交
152
    taosHashPut(pMgmt->dnodeHash, &pDnodeEp->id, sizeof(int32_t), pDnodeEp, sizeof(SDnodeEp));
S
Shengliang Guan 已提交
153 154
  }

S
Shengliang Guan 已提交
155
  dndPrintDnodes(pDnode);
S
Shengliang Guan 已提交
156 157
}

S
Shengliang Guan 已提交
158
static bool dndIsEpChanged(SDnode *pDnode, int32_t dnodeId, char *pEp) {
S
Shengliang Guan 已提交
159 160
  bool changed = false;

S
Shengliang Guan 已提交
161 162 163 164
  SDnodeMgmt *pMgmt = &pDnode->dmgmt;
  taosRLockLatch(&pMgmt->latch);

  SDnodeEp *pDnodeEp = taosHashGet(pMgmt->dnodeHash, &dnodeId, sizeof(int32_t));
S
Shengliang Guan 已提交
165 166
  if (pDnodeEp != NULL) {
    char epstr[TSDB_EP_LEN + 1];
H
Haojun Liao 已提交
167
    snprintf(epstr, TSDB_EP_LEN, "%s:%u", pDnodeEp->ep.fqdn, pDnodeEp->ep.port);
S
Shengliang Guan 已提交
168
    changed = strcmp(pEp, epstr) != 0;
S
Shengliang Guan 已提交
169 170
  }

S
Shengliang Guan 已提交
171
  taosRUnLockLatch(&pMgmt->latch);
S
Shengliang Guan 已提交
172 173 174
  return changed;
}

S
Shengliang Guan 已提交
175 176
static int32_t dndReadDnodes(SDnode *pDnode) {
  SDnodeMgmt *pMgmt = &pDnode->dmgmt;
S
Shengliang Guan 已提交
177

S
Shengliang Guan 已提交
178 179 180 181 182 183
  pMgmt->pDnodeEps = taosArrayInit(1, sizeof(SDnodeEp));
  if (pMgmt->pDnodeEps == NULL) {
    dError("failed to calloc dnodeEp array since %s", strerror(errno));
    goto PRASE_DNODE_OVER;
  }

S
Shengliang Guan 已提交
184
  int32_t code = TSDB_CODE_DND_DNODE_READ_FILE_ERROR;
S
Shengliang Guan 已提交
185
  int32_t len = 0;
S
Shengliang Guan 已提交
186
  int32_t maxLen = 256 * 1024;
S
Shengliang Guan 已提交
187 188 189
  char   *content = calloc(1, maxLen + 1);
  cJSON  *root = NULL;

190 191 192
  // fp = fopen(pMgmt->file, "r");
  TdFilePtr pFile = taosOpenFile(pMgmt->file, TD_FILE_READ);
  if (pFile == NULL) {
S
Shengliang Guan 已提交
193 194
    dDebug("file %s not exist", pMgmt->file);
    code = 0;
S
Shengliang Guan 已提交
195
    goto PRASE_DNODE_OVER;
S
Shengliang Guan 已提交
196 197
  }

198
  len = (int32_t)taosReadFile(pFile, content, maxLen);
S
Shengliang Guan 已提交
199
  if (len <= 0) {
S
Shengliang Guan 已提交
200
    dError("failed to read %s since content is null", pMgmt->file);
S
Shengliang Guan 已提交
201
    goto PRASE_DNODE_OVER;
S
Shengliang Guan 已提交
202 203 204 205 206
  }

  content[len] = 0;
  root = cJSON_Parse(content);
  if (root == NULL) {
S
Shengliang Guan 已提交
207
    dError("failed to read %s since invalid json format", pMgmt->file);
S
Shengliang Guan 已提交
208
    goto PRASE_DNODE_OVER;
S
Shengliang Guan 已提交
209 210 211
  }

  cJSON *dnodeId = cJSON_GetObjectItem(root, "dnodeId");
S
Shengliang Guan 已提交
212
  if (!dnodeId || dnodeId->type != cJSON_Number) {
S
Shengliang Guan 已提交
213
    dError("failed to read %s since dnodeId not found", pMgmt->file);
S
Shengliang Guan 已提交
214
    goto PRASE_DNODE_OVER;
S
Shengliang Guan 已提交
215
  }
S
Shengliang Guan 已提交
216
  pMgmt->dnodeId = dnodeId->valueint;
S
Shengliang Guan 已提交
217

S
Shengliang Guan 已提交
218
  cJSON *clusterId = cJSON_GetObjectItem(root, "clusterId");
S
Shengliang Guan 已提交
219
  if (!clusterId || clusterId->type != cJSON_String) {
S
Shengliang Guan 已提交
220
    dError("failed to read %s since clusterId not found", pMgmt->file);
S
Shengliang Guan 已提交
221
    goto PRASE_DNODE_OVER;
S
Shengliang Guan 已提交
222
  }
S
Shengliang Guan 已提交
223
  pMgmt->clusterId = atoll(clusterId->valuestring);
S
Shengliang Guan 已提交
224

S
Shengliang Guan 已提交
225
  cJSON *dropped = cJSON_GetObjectItem(root, "dropped");
S
Shengliang Guan 已提交
226
  if (!dropped || dropped->type != cJSON_Number) {
S
Shengliang Guan 已提交
227
    dError("failed to read %s since dropped not found", pMgmt->file);
S
Shengliang Guan 已提交
228
    goto PRASE_DNODE_OVER;
S
Shengliang Guan 已提交
229
  }
S
Shengliang Guan 已提交
230
  pMgmt->dropped = dropped->valueint;
S
Shengliang Guan 已提交
231

S
Shengliang Guan 已提交
232 233 234
  cJSON *dnodes = cJSON_GetObjectItem(root, "dnodes");
  if (!dnodes || dnodes->type != cJSON_Array) {
    dError("failed to read %s since dnodes not found", pMgmt->file);
S
Shengliang Guan 已提交
235
    goto PRASE_DNODE_OVER;
S
Shengliang Guan 已提交
236 237
  }

S
Shengliang Guan 已提交
238 239 240
  int32_t numOfDnodes = cJSON_GetArraySize(dnodes);
  if (numOfDnodes <= 0) {
    dError("failed to read %s since numOfDnodes:%d invalid", pMgmt->file, numOfDnodes);
S
Shengliang Guan 已提交
241
    goto PRASE_DNODE_OVER;
S
Shengliang Guan 已提交
242 243
  }

S
Shengliang Guan 已提交
244
  for (int32_t i = 0; i < numOfDnodes; ++i) {
S
Shengliang Guan 已提交
245 246
    cJSON *node = cJSON_GetArrayItem(dnodes, i);
    if (node == NULL) break;
S
Shengliang Guan 已提交
247

S
Shengliang Guan 已提交
248
    SDnodeEp dnodeEp = {0};
S
Shengliang Guan 已提交
249

H
Haojun Liao 已提交
250 251
    cJSON *did = cJSON_GetObjectItem(node, "id");
    if (!did || did->type != cJSON_Number) {
S
Shengliang Guan 已提交
252
      dError("failed to read %s since dnodeId not found", pMgmt->file);
S
Shengliang Guan 已提交
253
      goto PRASE_DNODE_OVER;
S
Shengliang Guan 已提交
254
    }
H
Haojun Liao 已提交
255

S
Shengliang Guan 已提交
256
    dnodeEp.id = dnodeId->valueint;
S
Shengliang Guan 已提交
257

S
Shengliang Guan 已提交
258
    cJSON *dnodeFqdn = cJSON_GetObjectItem(node, "fqdn");
S
Shengliang Guan 已提交
259
    if (!dnodeFqdn || dnodeFqdn->type != cJSON_String || dnodeFqdn->valuestring == NULL) {
S
Shengliang Guan 已提交
260
      dError("failed to read %s since dnodeFqdn not found", pMgmt->file);
S
Shengliang Guan 已提交
261
      goto PRASE_DNODE_OVER;
S
Shengliang Guan 已提交
262
    }
S
Shengliang Guan 已提交
263
    tstrncpy(dnodeEp.ep.fqdn, dnodeFqdn->valuestring, TSDB_FQDN_LEN);
S
Shengliang Guan 已提交
264

S
Shengliang Guan 已提交
265 266
    cJSON *dnodePort = cJSON_GetObjectItem(node, "port");
    if (!dnodePort || dnodePort->type != cJSON_Number) {
S
Shengliang Guan 已提交
267
      dError("failed to read %s since dnodePort not found", pMgmt->file);
S
Shengliang Guan 已提交
268
      goto PRASE_DNODE_OVER;
S
Shengliang Guan 已提交
269
    }
H
Haojun Liao 已提交
270

S
Shengliang Guan 已提交
271
    dnodeEp.ep.port = dnodePort->valueint;
S
Shengliang Guan 已提交
272 273 274

    cJSON *isMnode = cJSON_GetObjectItem(node, "isMnode");
    if (!isMnode || isMnode->type != cJSON_Number) {
S
Shengliang Guan 已提交
275
      dError("failed to read %s since isMnode not found", pMgmt->file);
S
Shengliang Guan 已提交
276 277
      goto PRASE_DNODE_OVER;
    }
S
Shengliang Guan 已提交
278 279 280
    dnodeEp.isMnode = isMnode->valueint;

    taosArrayPush(pMgmt->pDnodeEps, &dnodeEp);
S
Shengliang Guan 已提交
281 282
  }

S
Shengliang Guan 已提交
283 284 285
  code = 0;
  dInfo("succcessed to read file %s", pMgmt->file);
  dndPrintDnodes(pDnode);
S
Shengliang Guan 已提交
286

S
Shengliang Guan 已提交
287
PRASE_DNODE_OVER:
S
Shengliang Guan 已提交
288 289
  if (content != NULL) free(content);
  if (root != NULL) cJSON_Delete(root);
290
  if (pFile != NULL) taosCloseFile(&pFile);
S
Shengliang Guan 已提交
291

S
Shengliang Guan 已提交
292 293
  if (dndIsEpChanged(pDnode, pMgmt->dnodeId, pDnode->cfg.localEp)) {
    dError("localEp %s different with %s and need reconfigured", pDnode->cfg.localEp, pMgmt->file);
S
Shengliang Guan 已提交
294 295 296
    return -1;
  }

S
Shengliang Guan 已提交
297 298 299
  if (taosArrayGetSize(pMgmt->pDnodeEps) == 0) {
    SDnodeEp dnodeEp = {0};
    dnodeEp.isMnode = 1;
S
Shengliang Guan 已提交
300
    taosGetFqdnPortFromEp(pDnode->cfg.firstEp, &dnodeEp.ep);
S
Shengliang Guan 已提交
301
    taosArrayPush(pMgmt->pDnodeEps, &dnodeEp);
S
Shengliang Guan 已提交
302 303
  }

S
Shengliang Guan 已提交
304
  dndResetDnodes(pDnode, pMgmt->pDnodeEps);
S
Shengliang Guan 已提交
305 306 307 308 309

  terrno = 0;
  return 0;
}

S
Shengliang Guan 已提交
310 311
static int32_t dndWriteDnodes(SDnode *pDnode) {
  SDnodeMgmt *pMgmt = &pDnode->dmgmt;
S
Shengliang Guan 已提交
312

313 314 315
  // FILE *fp = fopen(pMgmt->file, "w");
  TdFilePtr pFile = taosOpenFile(pMgmt->file, TD_FILE_CTEATE | TD_FILE_WRITE | TD_FILE_TRUNC);
  if (pFile == NULL) {
S
Shengliang Guan 已提交
316 317
    dError("failed to write %s since %s", pMgmt->file, strerror(errno));
    terrno = TAOS_SYSTEM_ERROR(errno);
S
Shengliang Guan 已提交
318 319 320 321
    return -1;
  }

  int32_t len = 0;
S
Shengliang Guan 已提交
322
  int32_t maxLen = 256 * 1024;
S
Shengliang Guan 已提交
323 324 325
  char   *content = calloc(1, maxLen + 1);

  len += snprintf(content + len, maxLen - len, "{\n");
S
Shengliang Guan 已提交
326
  len += snprintf(content + len, maxLen - len, "  \"dnodeId\": %d,\n", pMgmt->dnodeId);
327
  len += snprintf(content + len, maxLen - len, "  \"clusterId\": \"%" PRId64 "\",\n", pMgmt->clusterId);
S
Shengliang Guan 已提交
328 329
  len += snprintf(content + len, maxLen - len, "  \"dropped\": %d,\n", pMgmt->dropped);
  len += snprintf(content + len, maxLen - len, "  \"dnodes\": [{\n");
S
Shengliang Guan 已提交
330 331 332 333

  int32_t numOfEps = (int32_t)taosArrayGetSize(pMgmt->pDnodeEps);
  for (int32_t i = 0; i < numOfEps; ++i) {
    SDnodeEp *pDnodeEp = taosArrayGet(pMgmt->pDnodeEps, i);
S
Shengliang Guan 已提交
334
    len += snprintf(content + len, maxLen - len, "    \"id\": %d,\n", pDnodeEp->id);
H
Haojun Liao 已提交
335 336
    len += snprintf(content + len, maxLen - len, "    \"fqdn\": \"%s\",\n", pDnodeEp->ep.fqdn);
    len += snprintf(content + len, maxLen - len, "    \"port\": %u,\n", pDnodeEp->ep.port);
S
Shengliang Guan 已提交
337
    len += snprintf(content + len, maxLen - len, "    \"isMnode\": %d\n", pDnodeEp->isMnode);
S
Shengliang Guan 已提交
338
    if (i < numOfEps - 1) {
S
Shengliang Guan 已提交
339 340 341 342 343 344 345
      len += snprintf(content + len, maxLen - len, "  },{\n");
    } else {
      len += snprintf(content + len, maxLen - len, "  }]\n");
    }
  }
  len += snprintf(content + len, maxLen - len, "}\n");

346 347 348
  taosWriteFile(pFile, content, len);
  taosFsyncFile(pFile);
  taosCloseFile(&pFile);
S
Shengliang Guan 已提交
349 350 351
  free(content);
  terrno = 0;

S
Shengliang Guan 已提交
352
  pMgmt->updateTime = taosGetTimestampMs();
S
Shengliang Guan 已提交
353
  dDebug("successed to write %s", pMgmt->file);
S
Shengliang Guan 已提交
354 355 356
  return 0;
}

S
Shengliang Guan 已提交
357
void dndSendStatusReq(SDnode *pDnode) {
S
Shengliang Guan 已提交
358
  SStatusReq req = {0};
S
Shengliang Guan 已提交
359

S
Shengliang Guan 已提交
360 361
  SDnodeMgmt *pMgmt = &pDnode->dmgmt;
  taosRLockLatch(&pMgmt->latch);
S
config  
Shengliang Guan 已提交
362
  req.sver = tsVersion;
S
Shengliang Guan 已提交
363 364 365 366 367
  req.dver = pMgmt->dver;
  req.dnodeId = pMgmt->dnodeId;
  req.clusterId = pMgmt->clusterId;
  req.rebootTime = pMgmt->rebootTime;
  req.updateTime = pMgmt->updateTime;
S
config  
Shengliang Guan 已提交
368
  req.numOfCores = tsNumOfCores;
S
Shengliang Guan 已提交
369 370 371
  req.numOfSupportVnodes = pDnode->cfg.numOfSupportVnodes;
  memcpy(req.dnodeEp, pDnode->cfg.localEp, TSDB_EP_LEN);

S
Shengliang Guan 已提交
372
  req.clusterCfg.statusInterval = tsStatusInterval;
S
Shengliang Guan 已提交
373
  req.clusterCfg.checkTime = 0;
S
Shengliang Guan 已提交
374
  char timestr[32] = "1970-01-01 00:00:00.00";
S
Shengliang Guan 已提交
375
  (void)taosParseTime(timestr, &req.clusterCfg.checkTime, (int32_t)strlen(timestr), TSDB_TIME_PRECISION_MILLI, 0);
S
os env  
Shengliang Guan 已提交
376 377 378
  memcpy(req.clusterCfg.timezone, tsTimezone, TD_TIMEZONE_LEN);
  memcpy(req.clusterCfg.locale, tsLocale, TD_LOCALE_LEN);
  memcpy(req.clusterCfg.charset, tsCharset, TD_LOCALE_LEN);
S
Shengliang Guan 已提交
379
  taosRUnLockLatch(&pMgmt->latch);
S
Shengliang Guan 已提交
380

S
Shengliang Guan 已提交
381 382 383
  req.pVloads = taosArrayInit(TSDB_MAX_VNODES, sizeof(SVnodeLoad));
  dndGetVnodeLoads(pDnode, req.pVloads);

S
Shengliang Guan 已提交
384
  int32_t contLen = tSerializeSStatusReq(NULL, 0, &req);
S
Shengliang Guan 已提交
385
  void   *pHead = rpcMallocCont(contLen);
S
Shengliang Guan 已提交
386
  tSerializeSStatusReq(pHead, contLen, &req);
S
Shengliang Guan 已提交
387
  taosArrayDestroy(req.pVloads);
S
Shengliang Guan 已提交
388

S
Shengliang Guan 已提交
389
  SRpcMsg rpcMsg = {.pCont = pHead, .contLen = contLen, .msgType = TDMT_MND_STATUS, .ahandle = (void *)9527};
390
  pMgmt->statusSent = 1;
S
Shengliang Guan 已提交
391

S
Shengliang Guan 已提交
392
  dTrace("pDnode:%p, send status req to mnode", pDnode);
S
Shengliang Guan 已提交
393
  dndSendReqToMnode(pDnode, &rpcMsg);
S
Shengliang Guan 已提交
394 395
}

S
Shengliang Guan 已提交
396 397
static void dndUpdateDnodeCfg(SDnode *pDnode, SDnodeCfg *pCfg) {
  SDnodeMgmt *pMgmt = &pDnode->dmgmt;
S
Shengliang Guan 已提交
398
  if (pMgmt->dnodeId == 0) {
S
Shengliang Guan 已提交
399
    dInfo("set dnodeId:%d clusterId:%" PRId64, pCfg->dnodeId, pCfg->clusterId);
S
Shengliang Guan 已提交
400
    taosWLockLatch(&pMgmt->latch);
S
Shengliang Guan 已提交
401 402
    pMgmt->dnodeId = pCfg->dnodeId;
    pMgmt->clusterId = pCfg->clusterId;
403
    dndWriteDnodes(pDnode);
S
Shengliang Guan 已提交
404
    taosWUnLockLatch(&pMgmt->latch);
S
Shengliang Guan 已提交
405
  }
S
Shengliang Guan 已提交
406 407
}

S
Shengliang Guan 已提交
408 409 410
static void dndUpdateDnodeEps(SDnode *pDnode, SArray *pDnodeEps) {
  int32_t numOfEps = taosArrayGetSize(pDnodeEps);
  if (numOfEps <= 0) return;
S
Shengliang Guan 已提交
411

S
Shengliang Guan 已提交
412 413
  SDnodeMgmt *pMgmt = &pDnode->dmgmt;
  taosWLockLatch(&pMgmt->latch);
S
Shengliang Guan 已提交
414

S
Shengliang Guan 已提交
415 416
  int32_t numOfEpsOld = (int32_t)taosArrayGetSize(pMgmt->pDnodeEps);
  if (numOfEps != numOfEpsOld) {
S
Shengliang Guan 已提交
417 418
    dndResetDnodes(pDnode, pDnodeEps);
    dndWriteDnodes(pDnode);
S
Shengliang Guan 已提交
419
  } else {
S
Shengliang Guan 已提交
420 421
    int32_t size = numOfEps * sizeof(SDnodeEp);
    if (memcmp(pMgmt->pDnodeEps->pData, pDnodeEps->pData, size) != 0) {
S
Shengliang Guan 已提交
422 423
      dndResetDnodes(pDnode, pDnodeEps);
      dndWriteDnodes(pDnode);
S
Shengliang Guan 已提交
424 425 426
    }
  }

S
Shengliang Guan 已提交
427
  taosWUnLockLatch(&pMgmt->latch);
S
Shengliang Guan 已提交
428 429
}

S
Shengliang Guan 已提交
430
static void dndProcessStatusRsp(SDnode *pDnode, SRpcMsg *pRsp) {
431 432
  SDnodeMgmt *pMgmt = &pDnode->dmgmt;

S
Shengliang Guan 已提交
433 434
  if (pRsp->code != TSDB_CODE_SUCCESS) {
    if (pRsp->code == TSDB_CODE_MND_DNODE_NOT_EXIST && !pMgmt->dropped && pMgmt->dnodeId > 0) {
S
Shengliang Guan 已提交
435 436 437 438
      dInfo("dnode:%d, set to dropped since not exist in mnode", pMgmt->dnodeId);
      pMgmt->dropped = 1;
      dndWriteDnodes(pDnode);
    }
S
Shengliang Guan 已提交
439 440
  } else {
    SStatusRsp statusRsp = {0};
S
Shengliang Guan 已提交
441 442
    if (pRsp->pCont != NULL && pRsp->contLen != 0 &&
        tDeserializeSStatusRsp(pRsp->pCont, pRsp->contLen, &statusRsp) == 0) {
S
Shengliang Guan 已提交
443 444 445
      pMgmt->dver = statusRsp.dver;
      dndUpdateDnodeCfg(pDnode, &statusRsp.dnodeCfg);
      dndUpdateDnodeEps(pDnode, statusRsp.pDnodeEps);
S
Shengliang Guan 已提交
446
    }
S
Shengliang Guan 已提交
447
    taosArrayDestroy(statusRsp.pDnodeEps);
S
Shengliang Guan 已提交
448
  }
S
Shengliang Guan 已提交
449

450
  pMgmt->statusSent = 0;
S
Shengliang Guan 已提交
451
}
S
Shengliang Guan 已提交
452

S
Shengliang Guan 已提交
453
static void dndProcessAuthRsp(SDnode *pDnode, SRpcMsg *pReq) { dError("auth rsp is received, but not supported yet"); }
S
Shengliang Guan 已提交
454

S
Shengliang Guan 已提交
455 456 457
static void dndProcessGrantRsp(SDnode *pDnode, SRpcMsg *pReq) {
  dError("grant rsp is received, but not supported yet");
}
S
Shengliang Guan 已提交
458

S
Shengliang Guan 已提交
459 460 461
static int32_t dndProcessConfigDnodeReq(SDnode *pDnode, SRpcMsg *pReq) {
  dError("config req is received, but not supported yet");
  SDCfgDnodeReq *pCfg = pReq->pCont;
S
Shengliang Guan 已提交
462
  return TSDB_CODE_OPS_NOT_SUPPORT;
S
Shengliang Guan 已提交
463 464
}

S
Shengliang Guan 已提交
465 466
void dndProcessStartupReq(SDnode *pDnode, SRpcMsg *pReq) {
  dDebug("startup req is received");
S
Shengliang Guan 已提交
467

S
Shengliang Guan 已提交
468
  SStartupReq *pStartup = rpcMallocCont(sizeof(SStartupReq));
S
Shengliang Guan 已提交
469
  dndGetStartup(pDnode, pStartup);
S
Shengliang Guan 已提交
470

S
Shengliang Guan 已提交
471
  dDebug("startup req is sent, step:%s desc:%s finished:%d", pStartup->name, pStartup->desc, pStartup->finished);
S
Shengliang Guan 已提交
472

S
Shengliang Guan 已提交
473
  SRpcMsg rpcRsp = {.handle = pReq->handle, .pCont = pStartup, .contLen = sizeof(SStartupReq)};
S
Shengliang Guan 已提交
474
  rpcSendResponse(&rpcRsp);
S
Shengliang Guan 已提交
475 476
}

S
Shengliang Guan 已提交
477
static void dndGetMonitorBasicInfo(SDnode *pDnode, SMonBasicInfo *pInfo) {
S
monitor  
Shengliang Guan 已提交
478 479 480 481
  pInfo->dnode_id = dndGetDnodeId(pDnode);
  tstrncpy(pInfo->dnode_ep, tsLocalEp, TSDB_EP_LEN);
}

S
Shengliang Guan 已提交
482 483 484 485 486 487 488
static void dndGetMonitorDnodeInfo(SDnode *pDnode, SMonDnodeInfo *pInfo) {
  pInfo->uptime = (taosGetTimestampMs() - pDnode->dmgmt.rebootTime) / (86400000.0f);
  taosGetCpuUsage(&pInfo->cpu_engine, &pInfo->cpu_system);
  pInfo->cpu_cores = tsNumOfCores;
  taosGetProcMemory(&pInfo->mem_engine);
  taosGetSysMemory(&pInfo->mem_system);
  pInfo->mem_total = tsTotalMemoryKB;
S
monitor  
Shengliang Guan 已提交
489
  pInfo->disk_engine = 0;
S
Shengliang Guan 已提交
490
  pInfo->disk_used = tsDataSpace.size.used;
S
Shengliang Guan 已提交
491 492 493
  pInfo->disk_total = tsDataSpace.size.total;
  taosGetBandSpeed(&pInfo->net_in, &pInfo->net_out);
  taosGetIOSpeed(&pInfo->io_read, &pInfo->io_write, &pInfo->io_read_disk, &pInfo->io_write_disk);
S
monitor  
Shengliang Guan 已提交
494 495 496 497 498 499 500 501
  pInfo->req_select = 0;
  pInfo->req_select_rate = 0;
  pInfo->req_insert = 0;
  pInfo->req_insert_success = 0;
  pInfo->req_insert_rate = 0;
  pInfo->req_insert_batch = 0;
  pInfo->req_insert_batch_success = 0;
  pInfo->req_insert_batch_rate = 0;
S
Shengliang Guan 已提交
502 503 504
  pInfo->errors = tsNumOfErrorLogs;
  pInfo->vnodes_num = pDnode->vmgmt.totalVnodes;
  pInfo->masters = pDnode->vmgmt.masterNum;
S
monitor  
Shengliang Guan 已提交
505
  pInfo->has_mnode = dndIsMnode(pDnode);
S
Shengliang Guan 已提交
506
}
S
monitor  
Shengliang Guan 已提交
507

S
monitor  
Shengliang Guan 已提交
508
static void dndSendMonitorReport(SDnode *pDnode) {
S
monitor  
Shengliang Guan 已提交
509 510 511
  if (!tsEnableMonitor || tsMonitorFqdn[0] == 0 || tsMonitorPort == 0) return;
  dTrace("pDnode:%p, send monitor report to %s:%u", pDnode, tsMonitorFqdn, tsMonitorPort);

S
monitor  
Shengliang Guan 已提交
512 513
  SMonInfo *pMonitor = monCreateMonitorInfo();
  if (pMonitor == NULL) return;
S
monitor  
Shengliang Guan 已提交
514

S
monitor  
Shengliang Guan 已提交
515
  SMonBasicInfo basicInfo = {0};
S
Shengliang Guan 已提交
516 517
  dndGetMonitorBasicInfo(pDnode, &basicInfo);
  monSetBasicInfo(pMonitor, &basicInfo);
S
monitor  
Shengliang Guan 已提交
518

S
monitor  
Shengliang Guan 已提交
519 520 521 522 523 524 525 526 527
  SMonClusterInfo clusterInfo = {0};
  SMonVgroupInfo  vgroupInfo = {0};
  SMonGrantInfo   grantInfo = {0};
  if (dndGetMnodeMonitorInfo(pDnode, &clusterInfo, &vgroupInfo, &grantInfo) == 0) {
    monSetClusterInfo(pMonitor, &clusterInfo);
    monSetVgroupInfo(pMonitor, &vgroupInfo);
    monSetGrantInfo(pMonitor, &grantInfo);
  }

S
monitor  
Shengliang Guan 已提交
528
  SMonDnodeInfo dnodeInfo = {0};
S
Shengliang Guan 已提交
529 530
  dndGetMonitorDnodeInfo(pDnode, &dnodeInfo);
  monSetDnodeInfo(pMonitor, &dnodeInfo);
S
monitor  
Shengliang Guan 已提交
531 532

  SMonDiskInfo diskInfo = {0};
S
Shengliang Guan 已提交
533
  if (dndGetMonitorDiskInfo(pDnode, &diskInfo) == 0) {
S
monitor  
Shengliang Guan 已提交
534 535 536
    monSetDiskInfo(pMonitor, &diskInfo);
  }

S
Shengliang Guan 已提交
537 538 539 540 541
  taosArrayDestroy(clusterInfo.dnodes);
  taosArrayDestroy(clusterInfo.mnodes);
  taosArrayDestroy(vgroupInfo.vgroups);
  taosArrayDestroy(diskInfo.datadirs);

S
monitor  
Shengliang Guan 已提交
542 543
  monSendReport(pMonitor);
  monCleanupMonitorInfo(pMonitor);
S
monitor  
Shengliang Guan 已提交
544 545
}

S
Shengliang Guan 已提交
546
static void *dnodeThreadRoutine(void *param) {
547 548
  SDnode     *pDnode = param;
  SDnodeMgmt *pMgmt = &pDnode->dmgmt;
S
monitor  
Shengliang Guan 已提交
549 550
  int64_t     lastStatusTime = taosGetTimestampMs();
  int64_t     lastMonitorTime = lastStatusTime;
S
Shengliang Guan 已提交
551

H
Haojun Liao 已提交
552 553
  setThreadName("dnode-hb");

S
Shengliang Guan 已提交
554 555
  while (true) {
    pthread_testcancel();
S
monitor  
Shengliang Guan 已提交
556 557 558 559
    taosMsleep(200);
    if (dndGetStat(pDnode) != DND_STAT_RUNNING || pMgmt->dropped) {
      continue;
    }
S
Shengliang Guan 已提交
560

S
monitor  
Shengliang Guan 已提交
561 562 563 564
    int64_t curTime = taosGetTimestampMs();

    float statusInterval = (curTime - lastStatusTime) / 1000.0f;
    if (statusInterval >= tsStatusInterval && !pMgmt->statusSent) {
S
Shengliang Guan 已提交
565
      dndSendStatusReq(pDnode);
S
monitor  
Shengliang Guan 已提交
566 567 568 569 570 571 572
      lastStatusTime = curTime;
    }

    float monitorInterval = (curTime - lastMonitorTime) / 1000.0f;
    if (monitorInterval >= tsMonitorInterval) {
      dndSendMonitorReport(pDnode);
      lastMonitorTime = curTime;
S
Shengliang Guan 已提交
573
    }
S
Shengliang Guan 已提交
574 575 576
  }
}

S
Shengliang Guan 已提交
577
int32_t dndInitMgmt(SDnode *pDnode) {
S
Shengliang Guan 已提交
578
  SDnodeMgmt *pMgmt = &pDnode->dmgmt;
S
Shengliang Guan 已提交
579

S
Shengliang Guan 已提交
580
  pMgmt->dnodeId = 0;
581
  pMgmt->rebootTime = taosGetTimestampMs();
S
Shengliang Guan 已提交
582 583
  pMgmt->dropped = 0;
  pMgmt->clusterId = 0;
S
Shengliang Guan 已提交
584
  taosInitRWLatch(&pMgmt->latch);
S
Shengliang Guan 已提交
585 586

  char path[PATH_MAX];
S
Shengliang Guan 已提交
587 588 589
  snprintf(path, PATH_MAX, "%s/dnode.json", pDnode->dir.dnode);
  pMgmt->file = strdup(path);
  if (pMgmt->file == NULL) {
S
Shengliang Guan 已提交
590 591 592 593
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return -1;
  }

594
  pMgmt->dnodeHash = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);
S
Shengliang Guan 已提交
595
  if (pMgmt->dnodeHash == NULL) {
S
Shengliang Guan 已提交
596
    dError("failed to init dnode hash");
S
Shengliang Guan 已提交
597 598
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return -1;
S
Shengliang Guan 已提交
599 600
  }

S
Shengliang Guan 已提交
601 602
  if (dndReadDnodes(pDnode) != 0) {
    dError("failed to read file:%s since %s", pMgmt->file, terrstr());
S
Shengliang Guan 已提交
603
    return -1;
S
Shengliang Guan 已提交
604 605
  }

S
Shengliang Guan 已提交
606
  if (pMgmt->dropped) {
S
Shengliang Guan 已提交
607
    dError("dnode not start since its already dropped");
S
Shengliang Guan 已提交
608 609 610
    return -1;
  }

S
Shengliang Guan 已提交
611 612
  if (dndInitWorker(pDnode, &pMgmt->mgmtWorker, DND_WORKER_SINGLE, "dnode-mgmt", 1, 1, dndProcessMgmtQueue) != 0) {
    dError("failed to start dnode mgmt worker since %s", terrstr());
S
Shengliang Guan 已提交
613 614
    return -1;
  }
S
Shengliang Guan 已提交
615

616 617 618 619 620
  if (dndInitWorker(pDnode, &pMgmt->statusWorker, DND_WORKER_SINGLE, "dnode-status", 1, 1, dndProcessMgmtQueue) != 0) {
    dError("failed to start dnode mgmt worker since %s", terrstr());
    return -1;
  }

S
Shengliang Guan 已提交
621 622
  pMgmt->threadId = taosCreateThread(dnodeThreadRoutine, pDnode);
  if (pMgmt->threadId == NULL) {
S
Shengliang Guan 已提交
623 624 625
    dError("failed to init dnode thread");
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return -1;
S
Shengliang Guan 已提交
626 627
  }

S
Shengliang Guan 已提交
628
  dInfo("dnode-mgmt is initialized");
S
Shengliang Guan 已提交
629 630 631
  return 0;
}

S
Shengliang Guan 已提交
632
void dndStopMgmt(SDnode *pDnode) {
S
Shengliang Guan 已提交
633
  SDnodeMgmt *pMgmt = &pDnode->dmgmt;
S
Shengliang Guan 已提交
634
  dndCleanupWorker(&pMgmt->mgmtWorker);
635
  dndCleanupWorker(&pMgmt->statusWorker);
S
Shengliang Guan 已提交
636

S
Shengliang Guan 已提交
637 638 639
  if (pMgmt->threadId != NULL) {
    taosDestoryThread(pMgmt->threadId);
    pMgmt->threadId = NULL;
S
Shengliang Guan 已提交
640
  }
S
Shengliang Guan 已提交
641
}
S
Shengliang Guan 已提交
642

S
Shengliang Guan 已提交
643 644
void dndCleanupMgmt(SDnode *pDnode) {
  SDnodeMgmt *pMgmt = &pDnode->dmgmt;
S
Shengliang Guan 已提交
645
  taosWLockLatch(&pMgmt->latch);
S
Shengliang Guan 已提交
646

S
Shengliang Guan 已提交
647 648 649
  if (pMgmt->pDnodeEps != NULL) {
    taosArrayDestroy(pMgmt->pDnodeEps);
    pMgmt->pDnodeEps = NULL;
S
Shengliang Guan 已提交
650
  }
S
Shengliang Guan 已提交
651

S
Shengliang Guan 已提交
652 653 654
  if (pMgmt->dnodeHash != NULL) {
    taosHashCleanup(pMgmt->dnodeHash);
    pMgmt->dnodeHash = NULL;
S
Shengliang Guan 已提交
655
  }
S
Shengliang Guan 已提交
656

S
Shengliang Guan 已提交
657 658 659
  if (pMgmt->file != NULL) {
    free(pMgmt->file);
    pMgmt->file = NULL;
S
Shengliang Guan 已提交
660 661
  }

S
Shengliang Guan 已提交
662
  taosWUnLockLatch(&pMgmt->latch);
S
Shengliang Guan 已提交
663
  dInfo("dnode-mgmt is cleaned up");
S
Shengliang Guan 已提交
664 665
}

S
Shengliang Guan 已提交
666
void dndProcessMgmtMsg(SDnode *pDnode, SRpcMsg *pMsg, SEpSet *pEpSet) {
S
Shengliang Guan 已提交
667 668
  SDnodeMgmt *pMgmt = &pDnode->dmgmt;

S
Shengliang Guan 已提交
669
  if (pEpSet && pEpSet->numOfEps > 0 && pMsg->msgType == TDMT_MND_STATUS_RSP) {
S
Shengliang Guan 已提交
670 671 672
    dndUpdateMnodeEpSet(pDnode, pEpSet);
  }

673 674 675 676 677 678
  SDnodeWorker *pWorker = &pMgmt->mgmtWorker;
  if (pMsg->msgType == TDMT_MND_STATUS_RSP) {
    pWorker = &pMgmt->statusWorker;
  }

  if (dndWriteMsgToWorker(pWorker, pMsg, sizeof(SRpcMsg)) != 0) {
S
Shengliang Guan 已提交
679 680
    if (pMsg->msgType & 1u) {
      SRpcMsg rsp = {.handle = pMsg->handle, .code = TSDB_CODE_OUT_OF_MEMORY};
S
Shengliang Guan 已提交
681 682
      rpcSendResponse(&rsp);
    }
S
Shengliang Guan 已提交
683
    rpcFreeCont(pMsg->pCont);
S
Shengliang Guan 已提交
684 685 686 687 688 689 690
    taosFreeQitem(pMsg);
  }
}

static void dndProcessMgmtQueue(SDnode *pDnode, SRpcMsg *pMsg) {
  int32_t code = 0;

S
Shengliang Guan 已提交
691
  switch (pMsg->msgType) {
S
Shengliang Guan 已提交
692 693 694 695 696 697 698 699 700
    case TDMT_DND_CREATE_MNODE:
      code = dndProcessCreateMnodeReq(pDnode, pMsg);
      break;
    case TDMT_DND_ALTER_MNODE:
      code = dndProcessAlterMnodeReq(pDnode, pMsg);
      break;
    case TDMT_DND_DROP_MNODE:
      code = dndProcessDropMnodeReq(pDnode, pMsg);
      break;
S
Shengliang Guan 已提交
701 702 703 704 705 706 707 708 709 710 711 712 713 714 715 716 717 718
    case TDMT_DND_CREATE_QNODE:
      code = dndProcessCreateQnodeReq(pDnode, pMsg);
      break;
    case TDMT_DND_DROP_QNODE:
      code = dndProcessDropQnodeReq(pDnode, pMsg);
      break;
    case TDMT_DND_CREATE_SNODE:
      code = dndProcessCreateSnodeReq(pDnode, pMsg);
      break;
    case TDMT_DND_DROP_SNODE:
      code = dndProcessDropSnodeReq(pDnode, pMsg);
      break;
    case TDMT_DND_CREATE_BNODE:
      code = dndProcessCreateBnodeReq(pDnode, pMsg);
      break;
    case TDMT_DND_DROP_BNODE:
      code = dndProcessDropBnodeReq(pDnode, pMsg);
      break;
H
Hongze Cheng 已提交
719
    case TDMT_DND_CONFIG_DNODE:
S
Shengliang Guan 已提交
720
      code = dndProcessConfigDnodeReq(pDnode, pMsg);
S
Shengliang Guan 已提交
721
      break;
H
Hongze Cheng 已提交
722
    case TDMT_MND_STATUS_RSP:
S
Shengliang Guan 已提交
723
      dndProcessStatusRsp(pDnode, pMsg);
S
Shengliang Guan 已提交
724
      break;
H
Hongze Cheng 已提交
725
    case TDMT_MND_AUTH_RSP:
S
Shengliang Guan 已提交
726
      dndProcessAuthRsp(pDnode, pMsg);
S
Shengliang Guan 已提交
727
      break;
H
Hongze Cheng 已提交
728
    case TDMT_MND_GRANT_RSP:
S
Shengliang Guan 已提交
729
      dndProcessGrantRsp(pDnode, pMsg);
S
Shengliang Guan 已提交
730
      break;
S
Shengliang Guan 已提交
731 732 733 734 735 736 737 738 739 740 741 742 743 744 745
    case TDMT_DND_CREATE_VNODE:
      code = dndProcessCreateVnodeReq(pDnode, pMsg);
      break;
    case TDMT_DND_ALTER_VNODE:
      code = dndProcessAlterVnodeReq(pDnode, pMsg);
      break;
    case TDMT_DND_DROP_VNODE:
      code = dndProcessDropVnodeReq(pDnode, pMsg);
      break;
    case TDMT_DND_SYNC_VNODE:
      code = dndProcessSyncVnodeReq(pDnode, pMsg);
      break;
    case TDMT_DND_COMPACT_VNODE:
      code = dndProcessCompactVnodeReq(pDnode, pMsg);
      break;
S
Shengliang Guan 已提交
746
    default:
S
Shengliang Guan 已提交
747 748
      terrno = TSDB_CODE_MSG_NOT_PROCESSED;
      code = -1;
S
Shengliang Guan 已提交
749
      dError("RPC %p, dnode msg:%s not processed", pMsg->handle, TMSG_INFO(pMsg->msgType));
S
Shengliang Guan 已提交
750 751 752 753 754 755 756
      break;
  }

  if (pMsg->msgType & 1u) {
    if (code != 0) code = terrno;
    SRpcMsg rsp = {.code = code, .handle = pMsg->handle, .ahandle = pMsg->ahandle};
    rpcSendResponse(&rsp);
S
Shengliang Guan 已提交
757
  }
S
Shengliang Guan 已提交
758

S
Shengliang Guan 已提交
759
  rpcFreeCont(pMsg->pCont);
S
Shengliang Guan 已提交
760
  pMsg->pCont = NULL;
S
Shengliang Guan 已提交
761
  taosFreeQitem(pMsg);
S
Shengliang Guan 已提交
762
}