dmHandle.c 9.5 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
Shengliang Guan 已提交
17
#include "dmImp.h"
S
shm  
Shengliang Guan 已提交
18

S
Shengliang Guan 已提交
19
static void dmUpdateDnodeCfg(SDnode *pDnode, SDnodeCfg *pCfg) {
20
  if (pDnode->data.dnodeId == 0 || pDnode->data.clusterId == 0) {
S
Shengliang Guan 已提交
21 22 23 24 25 26 27 28
    dInfo("set dnodeId:%d clusterId:%" PRId64, pCfg->dnodeId, pCfg->clusterId);
    taosWLockLatch(&pDnode->data.latch);
    pDnode->data.dnodeId = pCfg->dnodeId;
    pDnode->data.clusterId = pCfg->clusterId;
    dmWriteEps(pDnode);
    taosWUnLockLatch(&pDnode->data.latch);
  }
}
S
Shengliang Guan 已提交
29

S
Shengliang Guan 已提交
30
static int32_t dmProcessStatusRsp(SDnode *pDnode, SRpcMsg *pRsp) {
S
Shengliang Guan 已提交
31 32 33 34
  if (pRsp->code != TSDB_CODE_SUCCESS) {
    if (pRsp->code == TSDB_CODE_MND_DNODE_NOT_EXIST && !pDnode->data.dropped && pDnode->data.dnodeId > 0) {
      dInfo("dnode:%d, set to dropped since not exist in mnode", pDnode->data.dnodeId);
      pDnode->data.dropped = 1;
S
Shengliang Guan 已提交
35
      dmWriteEps(pDnode);
S
Shengliang Guan 已提交
36 37 38
    }
  } else {
    SStatusRsp statusRsp = {0};
S
Shengliang Guan 已提交
39 40
    if (pRsp->pCont != NULL && pRsp->contLen > 0 &&
        tDeserializeSStatusRsp(pRsp->pCont, pRsp->contLen, &statusRsp) == 0) {
S
Shengliang Guan 已提交
41 42 43
      pDnode->data.dnodeVer = statusRsp.dnodeVer;
      dmUpdateDnodeCfg(pDnode, &statusRsp.dnodeCfg);
      dmUpdateEps(pDnode, statusRsp.pDnodeEps);
S
Shengliang Guan 已提交
44
    }
dengyihao's avatar
dengyihao 已提交
45
    rpcFreeCont(pRsp->pCont);
S
Shengliang Guan 已提交
46 47 48 49 50 51 52
    tFreeSStatusRsp(&statusRsp);
  }

  return TSDB_CODE_SUCCESS;
}

void dmSendStatusReq(SDnode *pDnode) {
S
shm  
Shengliang Guan 已提交
53 54
  SStatusReq req = {0};

S
Shengliang Guan 已提交
55
  taosRLockLatch(&pDnode->data.latch);
S
shm  
Shengliang Guan 已提交
56
  req.sver = tsVersion;
S
Shengliang Guan 已提交
57
  req.dnodeVer = pDnode->data.dnodeVer;
S
Shengliang Guan 已提交
58 59
  req.dnodeId = pDnode->data.dnodeId;
  req.clusterId = pDnode->data.clusterId;
60
  if (req.clusterId == 0) req.dnodeId = 0;
S
Shengliang Guan 已提交
61
  req.rebootTime = pDnode->data.rebootTime;
S
Shengliang Guan 已提交
62
  req.updateTime = pDnode->data.updateTime;
S
shm  
Shengliang Guan 已提交
63
  req.numOfCores = tsNumOfCores;
S
Shengliang Guan 已提交
64 65
  req.numOfSupportVnodes = pDnode->data.supportVnodes;
  tstrncpy(req.dnodeEp, pDnode->data.localEp, TSDB_EP_LEN);
S
shm  
Shengliang Guan 已提交
66 67 68 69 70

  req.clusterCfg.statusInterval = tsStatusInterval;
  req.clusterCfg.checkTime = 0;
  char timestr[32] = "1970-01-01 00:00:00.00";
  (void)taosParseTime(timestr, &req.clusterCfg.checkTime, (int32_t)strlen(timestr), TSDB_TIME_PRECISION_MILLI, 0);
wafwerar's avatar
wafwerar 已提交
71
  memcpy(req.clusterCfg.timezone, tsTimezoneStr, TD_TIMEZONE_LEN);
S
shm  
Shengliang Guan 已提交
72 73
  memcpy(req.clusterCfg.locale, tsLocale, TD_LOCALE_LEN);
  memcpy(req.clusterCfg.charset, tsCharset, TD_LOCALE_LEN);
S
Shengliang Guan 已提交
74
  taosRUnLockLatch(&pDnode->data.latch);
S
shm  
Shengliang Guan 已提交
75

S
Shengliang Guan 已提交
76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92
  SMonVloadInfo vinfo = {0};
  dmGetVnodeLoads(pDnode, &vinfo);
  req.pVloads = vinfo.pVloads;
  pDnode->data.unsyncedVgId = 0;
  pDnode->data.vndState = TAOS_SYNC_STATE_LEADER;
  for (int32_t i = 0; i < taosArrayGetSize(req.pVloads); ++i) {
    SVnodeLoad *pLoad = taosArrayGet(req.pVloads, i);
    if (pLoad->syncState != TAOS_SYNC_STATE_LEADER && pLoad->syncState != TAOS_SYNC_STATE_FOLLOWER) {
      pDnode->data.unsyncedVgId = pLoad->vgId;
      pDnode->data.vndState = pLoad->syncState;
    }
  }

  SMonMloadInfo minfo = {0};
  dmGetMnodeLoads(pDnode, &minfo);
  pDnode->data.isMnode = minfo.isMnode;
  pDnode->data.mndState = minfo.load.syncState;
S
shm  
Shengliang Guan 已提交
93 94

  int32_t contLen = tSerializeSStatusReq(NULL, 0, &req);
S
Shengliang Guan 已提交
95
  void   *pHead = rpcMallocCont(contLen);
S
shm  
Shengliang Guan 已提交
96
  tSerializeSStatusReq(pHead, contLen, &req);
S
Shengliang Guan 已提交
97
  tFreeSStatusReq(&req);
S
shm  
Shengliang Guan 已提交
98

S
shm  
Shengliang Guan 已提交
99
  SRpcMsg rpcMsg = {.pCont = pHead, .contLen = contLen, .msgType = TDMT_MND_STATUS, .ahandle = (void *)0x9527};
S
Shengliang Guan 已提交
100 101
  SRpcMsg rpcRsp = {0};

S
shm  
Shengliang Guan 已提交
102
  dTrace("send req:%s to mnode, app:%p", TMSG_INFO(rpcMsg.msgType), rpcMsg.ahandle);
S
Shengliang Guan 已提交
103
  dmSendToMnodeRecv(pDnode, &rpcMsg, &rpcRsp);
S
Shengliang Guan 已提交
104
  dmProcessStatusRsp(pDnode, &rpcRsp);
S
shm  
Shengliang Guan 已提交
105 106
}

S
Shengliang Guan 已提交
107
int32_t dmProcessAuthRsp(SDnode *pDnode, SNodeMsg *pMsg) {
S
shm  
Shengliang Guan 已提交
108 109 110 111
  SRpcMsg *pRsp = &pMsg->rpcMsg;
  dError("auth rsp is received, but not supported yet");
  return 0;
}
S
shm  
Shengliang Guan 已提交
112

S
Shengliang Guan 已提交
113
int32_t dmProcessGrantRsp(SDnode *pDnode, SNodeMsg *pMsg) {
S
shm  
Shengliang Guan 已提交
114 115 116 117
  SRpcMsg *pRsp = &pMsg->rpcMsg;
  dError("grant rsp is received, but not supported yet");
  return 0;
}
S
shm  
Shengliang Guan 已提交
118

S
Shengliang Guan 已提交
119
int32_t dmProcessConfigReq(SDnode *pDnode, SNodeMsg *pMsg) {
S
Shengliang Guan 已提交
120
  SRpcMsg       *pReq = &pMsg->rpcMsg;
S
shm  
Shengliang Guan 已提交
121
  SDCfgDnodeReq *pCfg = pReq->pCont;
S
shm  
Shengliang Guan 已提交
122
  dError("config req is received, but not supported yet");
S
shm  
Shengliang Guan 已提交
123 124 125
  return TSDB_CODE_OPS_NOT_SUPPORT;
}

S
Shengliang Guan 已提交
126 127
int32_t dmProcessCreateNodeReq(SDnode *pDnode, EDndNodeType ntype, SNodeMsg *pMsg) {
  SMgmtWrapper *pWrapper = dmAcquireWrapper(pDnode, ntype);
S
Shengliang Guan 已提交
128
  if (pWrapper != NULL) {
S
Shengliang Guan 已提交
129
    dmReleaseWrapper(pWrapper);
S
Shengliang Guan 已提交
130 131 132 133 134
    terrno = TSDB_CODE_NODE_ALREADY_DEPLOYED;
    dError("failed to create node since %s", terrstr());
    return -1;
  }

135
  taosThreadMutexLock(&pDnode->mutex);
S
Shengliang Guan 已提交
136 137 138 139 140 141 142 143
  pWrapper = &pDnode->wrappers[ntype];

  if (taosMkDir(pWrapper->path) != 0) {
    terrno = TAOS_SYSTEM_ERROR(errno);
    dError("failed to create dir:%s since %s", pWrapper->path, terrstr());
    return -1;
  }

S
Shengliang Guan 已提交
144
  int32_t code = (*pWrapper->fp.createFp)(pWrapper, pMsg);
S
Shengliang Guan 已提交
145
  if (code != 0) {
S
Shengliang Guan 已提交
146
    dError("node:%s, failed to create since %s", pWrapper->name, terrstr());
S
Shengliang Guan 已提交
147
  } else {
S
Shengliang Guan 已提交
148
    dDebug("node:%s, has been created", pWrapper->name);
149
    (void)dmOpenNode(pWrapper);
S
Shengliang Guan 已提交
150
    pWrapper->required = true;
S
Shengliang Guan 已提交
151
    pWrapper->deployed = true;
152
    pWrapper->procType = pDnode->ptype;
S
Shengliang Guan 已提交
153 154
  }

155
  taosThreadMutexUnlock(&pDnode->mutex);
S
Shengliang Guan 已提交
156 157 158
  return code;
}

S
Shengliang Guan 已提交
159 160
int32_t dmProcessDropNodeReq(SDnode *pDnode, EDndNodeType ntype, SNodeMsg *pMsg) {
  SMgmtWrapper *pWrapper = dmAcquireWrapper(pDnode, ntype);
S
Shengliang Guan 已提交
161 162 163 164 165 166
  if (pWrapper == NULL) {
    terrno = TSDB_CODE_NODE_NOT_DEPLOYED;
    dError("failed to drop node since %s", terrstr());
    return -1;
  }

167
  taosThreadMutexLock(&pDnode->mutex);
S
Shengliang Guan 已提交
168

S
Shengliang Guan 已提交
169
  int32_t code = (*pWrapper->fp.dropFp)(pWrapper, pMsg);
S
Shengliang Guan 已提交
170 171 172 173
  if (code != 0) {
    dError("node:%s, failed to drop since %s", pWrapper->name, terrstr());
  } else {
    dDebug("node:%s, has been dropped", pWrapper->name);
174 175
    pWrapper->required = false;
    pWrapper->deployed = false;
S
Shengliang Guan 已提交
176 177
  }

S
Shengliang Guan 已提交
178
  dmReleaseWrapper(pWrapper);
S
Shengliang Guan 已提交
179 180

  if (code == 0) {
181
    dmCloseNode(pWrapper);
182
    taosRemoveDir(pWrapper->path);
S
Shengliang Guan 已提交
183
  }
184
  taosThreadMutexUnlock(&pDnode->mutex);
S
Shengliang Guan 已提交
185
  return code;
S
Shengliang Guan 已提交
186 187
}

S
Shengliang Guan 已提交
188
static void dmSetMgmtMsgHandle(SMgmtWrapper *pWrapper) {
S
shm  
Shengliang Guan 已提交
189
  // Requests handled by DNODE
S
Shengliang Guan 已提交
190 191 192 193 194 195 196 197 198
  dmSetMsgHandle(pWrapper, TDMT_DND_CREATE_MNODE, dmProcessMgmtMsg, DEFAULT_HANDLE);
  dmSetMsgHandle(pWrapper, TDMT_DND_DROP_MNODE, dmProcessMgmtMsg, DEFAULT_HANDLE);
  dmSetMsgHandle(pWrapper, TDMT_DND_CREATE_QNODE, dmProcessMgmtMsg, DEFAULT_HANDLE);
  dmSetMsgHandle(pWrapper, TDMT_DND_DROP_QNODE, dmProcessMgmtMsg, DEFAULT_HANDLE);
  dmSetMsgHandle(pWrapper, TDMT_DND_CREATE_SNODE, dmProcessMgmtMsg, DEFAULT_HANDLE);
  dmSetMsgHandle(pWrapper, TDMT_DND_DROP_SNODE, dmProcessMgmtMsg, DEFAULT_HANDLE);
  dmSetMsgHandle(pWrapper, TDMT_DND_CREATE_BNODE, dmProcessMgmtMsg, DEFAULT_HANDLE);
  dmSetMsgHandle(pWrapper, TDMT_DND_DROP_BNODE, dmProcessMgmtMsg, DEFAULT_HANDLE);
  dmSetMsgHandle(pWrapper, TDMT_DND_CONFIG_DNODE, dmProcessMgmtMsg, DEFAULT_HANDLE);
S
shm  
Shengliang Guan 已提交
199 200

  // Requests handled by MNODE
S
Shengliang Guan 已提交
201 202
  dmSetMsgHandle(pWrapper, TDMT_MND_GRANT_RSP, dmProcessMgmtMsg, DEFAULT_HANDLE);
  dmSetMsgHandle(pWrapper, TDMT_MND_AUTH_RSP, dmProcessMgmtMsg, DEFAULT_HANDLE);
S
shm  
Shengliang Guan 已提交
203
}
S
Shengliang Guan 已提交
204

S
Shengliang Guan 已提交
205 206 207 208 209 210 211 212 213
static int32_t dmStartMgmt(SMgmtWrapper *pWrapper) {
  if (dmStartStatusThread(pWrapper->pDnode) != 0) {
    return -1;
  }
  if (dmStartMonitorThread(pWrapper->pDnode) != 0) {
    return -1;
  }
  return 0;
}
S
Shengliang Guan 已提交
214

S
Shengliang Guan 已提交
215 216 217 218
static void dmStopMgmt(SMgmtWrapper *pWrapper) {
  dmStopMonitorThread(pWrapper->pDnode);
  dmStopStatusThread(pWrapper->pDnode);
}
S
Shengliang Guan 已提交
219

S
Shengliang Guan 已提交
220
static int32_t dmInitMgmt(SMgmtWrapper *pWrapper) {
S
Shengliang Guan 已提交
221
  dInfo("dnode-mgmt start to init");
S
Shengliang Guan 已提交
222 223 224 225 226 227 228 229 230
  SDnode *pDnode = pWrapper->pDnode;

  pDnode->data.dnodeHash = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_NO_LOCK);
  if (pDnode->data.dnodeHash == NULL) {
    dError("failed to init dnode hash");
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return -1;
  }

S
Shengliang Guan 已提交
231
  if (dmReadEps(pDnode) != 0) {
S
Shengliang Guan 已提交
232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248
    dError("failed to read file since %s", terrstr());
    return -1;
  }

  if (pDnode->data.dropped) {
    dError("dnode will not start since its already dropped");
    return -1;
  }

  if (dmStartWorker(pDnode) != 0) {
    return -1;
  }

  if (dmInitTrans(pDnode) != 0) {
    dError("failed to init transport since %s", terrstr());
    return -1;
  }
S
Shengliang Guan 已提交
249
  dmReportStartup(pDnode, "dnode-transport", "initialized");
S
Shengliang Guan 已提交
250

251
  if (udfStartUdfd(pDnode->data.dnodeId) != 0) {
S
slzhou 已提交
252 253
    dError("failed to start udfd");
  }
S
slzhou 已提交
254

S
Shengliang Guan 已提交
255
  dInfo("dnode-mgmt is initialized");
S
Shengliang Guan 已提交
256 257 258
  return 0;
}

S
Shengliang Guan 已提交
259
static void dmCleanupMgmt(SMgmtWrapper *pWrapper) {
S
Shengliang Guan 已提交
260
  dInfo("dnode-mgmt start to clean up");
S
Shengliang Guan 已提交
261
  SDnode *pDnode = pWrapper->pDnode;
262 263 264

  udfStopUdfd();

S
Shengliang Guan 已提交
265 266 267
  dmStopWorker(pDnode);

  taosWLockLatch(&pDnode->data.latch);
S
Shengliang Guan 已提交
268 269 270
  if (pDnode->data.dnodeEps != NULL) {
    taosArrayDestroy(pDnode->data.dnodeEps);
    pDnode->data.dnodeEps = NULL;
S
Shengliang Guan 已提交
271
  }
S
Shengliang Guan 已提交
272 273 274
  if (pDnode->data.dnodeHash != NULL) {
    taosHashCleanup(pDnode->data.dnodeHash);
    pDnode->data.dnodeHash = NULL;
S
Shengliang Guan 已提交
275 276 277
  }
  taosWUnLockLatch(&pDnode->data.latch);

S
Shengliang Guan 已提交
278
  dmCleanupTrans(pDnode);
S
Shengliang Guan 已提交
279
  dInfo("dnode-mgmt is cleaned up");
S
Shengliang Guan 已提交
280 281
}

S
Shengliang Guan 已提交
282
static int32_t dmRequireMgmt(SMgmtWrapper *pWrapper, bool *required) {
S
Shengliang Guan 已提交
283 284 285 286 287 288
  *required = true;
  return 0;
}

void dmSetMgmtFp(SMgmtWrapper *pWrapper) {
  SMgmtFp mgmtFp = {0};
S
Shengliang Guan 已提交
289 290 291 292 293
  mgmtFp.openFp = dmInitMgmt;
  mgmtFp.closeFp = dmCleanupMgmt;
  mgmtFp.startFp = dmStartMgmt;
  mgmtFp.stopFp = dmStopMgmt;
  mgmtFp.requiredFp = dmRequireMgmt;
S
Shengliang Guan 已提交
294

S
Shengliang Guan 已提交
295
  dmSetMgmtMsgHandle(pWrapper);
S
Shengliang Guan 已提交
296 297 298
  pWrapper->name = "dnode";
  pWrapper->fp = mgmtFp;
}