dmHandle.c 12.4 KB
Newer Older
S
Shengliang Guan 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17
/*
 * 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
#include "dmInt.h"
D
dapan1121 已提交
18
#include "systable.h"
wafwerar's avatar
wafwerar 已提交
19
#include "tgrant.h"
D
dapan1121 已提交
20 21

extern SConfig *tsCfg;
S
Shengliang Guan 已提交
22 23

static void dmUpdateDnodeCfg(SDnodeMgmt *pMgmt, SDnodeCfg *pCfg) {
S
Shengliang Guan 已提交
24
  if (pMgmt->pData->dnodeId == 0 || pMgmt->pData->clusterId == 0) {
S
Shengliang Guan 已提交
25
    dInfo("set local info, dnodeId:%d clusterId:%" PRId64, pCfg->dnodeId, pCfg->clusterId);
26
    taosThreadRwlockWrlock(&pMgmt->pData->lock);
S
Shengliang Guan 已提交
27 28 29
    pMgmt->pData->dnodeId = pCfg->dnodeId;
    pMgmt->pData->clusterId = pCfg->clusterId;
    dmWriteEps(pMgmt->pData);
30
    taosThreadRwlockUnlock(&pMgmt->pData->lock);
S
Shengliang Guan 已提交
31 32 33
  }
}

S
Shengliang Guan 已提交
34
static void dmProcessStatusRsp(SDnodeMgmt *pMgmt, SRpcMsg *pRsp) {
35
  const STraceId *trace = &pRsp->info.traceId;
36
  dGTrace("status rsp received from mnode, statusSeq:%d code:0x%x", pMgmt->statusSeq, pRsp->code);
37

S
Shengliang Guan 已提交
38
  if (pRsp->code != 0) {
S
Shengliang Guan 已提交
39
    if (pRsp->code == TSDB_CODE_MND_DNODE_NOT_EXIST && !pMgmt->pData->dropped && pMgmt->pData->dnodeId > 0) {
40 41
      dGInfo("dnode:%d, set to dropped since not exist in mnode, statusSeq:%d", pMgmt->pData->dnodeId,
             pMgmt->statusSeq);
S
Shengliang Guan 已提交
42 43
      pMgmt->pData->dropped = 1;
      dmWriteEps(pMgmt->pData);
44
      dInfo("dnode will exit since it is in the dropped state");
S
Shengliang Guan 已提交
45
      raise(SIGINT);
S
Shengliang Guan 已提交
46 47 48 49 50
    }
  } else {
    SStatusRsp statusRsp = {0};
    if (pRsp->pCont != NULL && pRsp->contLen > 0 &&
        tDeserializeSStatusRsp(pRsp->pCont, pRsp->contLen, &statusRsp) == 0) {
51
      if (pMgmt->pData->dnodeVer != statusRsp.dnodeVer) {
52
        dGInfo("status rsp received from mnode, statusSeq:%d:%d dnodeVer:%" PRId64 ":%" PRId64, pMgmt->statusSeq,
53
               statusRsp.statusSeq, pMgmt->pData->dnodeVer, statusRsp.dnodeVer);
54 55 56 57
        pMgmt->pData->dnodeVer = statusRsp.dnodeVer;
        dmUpdateDnodeCfg(pMgmt, &statusRsp.dnodeCfg);
        dmUpdateEps(pMgmt->pData, statusRsp.pDnodeEps);
      }
S
Shengliang Guan 已提交
58 59 60
    }
    tFreeSStatusRsp(&statusRsp);
  }
dengyihao's avatar
dengyihao 已提交
61
  rpcFreeCont(pRsp->pCont);
S
Shengliang Guan 已提交
62 63 64 65 66
}

void dmSendStatusReq(SDnodeMgmt *pMgmt) {
  SStatusReq req = {0};

67
  taosThreadRwlockRdlock(&pMgmt->pData->lock);
S
Shengliang Guan 已提交
68
  req.sver = tsVersion;
S
Shengliang Guan 已提交
69 70 71
  req.dnodeVer = pMgmt->pData->dnodeVer;
  req.dnodeId = pMgmt->pData->dnodeId;
  req.clusterId = pMgmt->pData->clusterId;
S
Shengliang Guan 已提交
72
  if (req.clusterId == 0) req.dnodeId = 0;
S
Shengliang Guan 已提交
73 74
  req.rebootTime = pMgmt->pData->rebootTime;
  req.updateTime = pMgmt->pData->updateTime;
S
Shengliang Guan 已提交
75
  req.numOfCores = tsNumOfCores;
76
  req.numOfSupportVnodes = tsNumOfSupportVnodes;
77 78
  req.memTotal = tsTotalMemoryKB * 1024;
  req.memAvail = req.memTotal - tsRpcQueueMemoryAllowed - 16 * 1024 * 1024;
79
  tstrncpy(req.dnodeEp, tsLocalEp, TSDB_EP_LEN);
S
Shengliang Guan 已提交
80 81 82 83 84 85 86 87

  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);
  memcpy(req.clusterCfg.timezone, tsTimezoneStr, TD_TIMEZONE_LEN);
  memcpy(req.clusterCfg.locale, tsLocale, TD_LOCALE_LEN);
  memcpy(req.clusterCfg.charset, tsCharset, TD_LOCALE_LEN);
88
  taosThreadRwlockUnlock(&pMgmt->pData->lock);
S
Shengliang Guan 已提交
89 90

  SMonVloadInfo vinfo = {0};
S
Shengliang Guan 已提交
91
  (*pMgmt->getVnodeLoadsFp)(&vinfo);
S
Shengliang Guan 已提交
92 93
  req.pVloads = vinfo.pVloads;

94
  SMonMloadInfo minfo = {0};
S
Shengliang Guan 已提交
95
  (*pMgmt->getMnodeLoadsFp)(&minfo);
96
  req.mload = minfo.load;
S
Shengliang Guan 已提交
97

D
dapan1121 已提交
98 99
  (*pMgmt->getQnodeLoadsFp)(&req.qload);

100 101 102
  pMgmt->statusSeq++;
  req.statusSeq = pMgmt->statusSeq;

S
Shengliang Guan 已提交
103 104 105 106 107
  int32_t contLen = tSerializeSStatusReq(NULL, 0, &req);
  void   *pHead = rpcMallocCont(contLen);
  tSerializeSStatusReq(pHead, contLen, &req);
  tFreeSStatusReq(&req);

dengyihao's avatar
dengyihao 已提交
108 109 110 111 112 113
  SRpcMsg rpcMsg = {.pCont = pHead,
                    .contLen = contLen,
                    .msgType = TDMT_MND_STATUS,
                    .info.ahandle = (void *)0x9527,
                    .info.refId = 0,
                    .info.noResp = 0};
S
Shengliang Guan 已提交
114 115
  SRpcMsg rpcRsp = {0};

116
  dTrace("send status req to mnode, dnodeVer:%" PRId64 " statusSeq:%d", req.dnodeVer, req.statusSeq);
S
Shengliang Guan 已提交
117 118 119 120

  SEpSet epSet = {0};
  dmGetMnodeEpSet(pMgmt->pData, &epSet);
  rpcSendRecv(pMgmt->msgCb.clientRpc, &epSet, &rpcMsg, &rpcRsp);
S
Shengliang Guan 已提交
121
  if (rpcRsp.code != 0) {
122
    dError("failed to send status req since %s, numOfEps:%d inUse:%d", tstrerror(rpcRsp.code), epSet.numOfEps,
S
Shengliang Guan 已提交
123 124 125 126
           epSet.inUse);
    for (int32_t i = 0; i < epSet.numOfEps; ++i) {
      dDebug("index:%d, mnode ep:%s:%u", i, epSet.eps[i].fqdn, epSet.eps[i].port);
    }
S
Shengliang Guan 已提交
127
  }
S
Shengliang Guan 已提交
128
  dmProcessStatusRsp(pMgmt, &rpcRsp);
S
Shengliang Guan 已提交
129 130
}

S
Shengliang Guan 已提交
131
int32_t dmProcessAuthRsp(SDnodeMgmt *pMgmt, SRpcMsg *pMsg) {
S
Shengliang Guan 已提交
132 133 134 135
  dError("auth rsp is received, but not supported yet");
  return 0;
}

S
Shengliang Guan 已提交
136
int32_t dmProcessGrantRsp(SDnodeMgmt *pMgmt, SRpcMsg *pMsg) {
S
Shengliang Guan 已提交
137 138 139 140
  dError("grant rsp is received, but not supported yet");
  return 0;
}

S
Shengliang Guan 已提交
141
int32_t dmProcessConfigReq(SDnodeMgmt *pMgmt, SRpcMsg *pMsg) {
S
Shengliang Guan 已提交
142
  SDCfgDnodeReq cfgReq = {0};
S
Shengliang Guan 已提交
143
  if (tDeserializeSDCfgDnodeReq(pMsg->pCont, pMsg->contLen, &cfgReq) != 0) {
S
Shengliang Guan 已提交
144 145 146 147 148 149 150
    terrno = TSDB_CODE_INVALID_MSG;
    return -1;
  }

  dInfo("start to config, option:%s, value:%s", cfgReq.config, cfgReq.value);
  taosCfgDynamicOptions(cfgReq.config, cfgReq.value);
  return 0;
S
Shengliang Guan 已提交
151 152
}

153 154 155 156 157 158
static void dmGetServerRunStatus(SDnodeMgmt *pMgmt, SServerStatusRsp *pStatus) {
  pStatus->statusCode = TSDB_SRV_STATUS_SERVICE_OK;
  pStatus->details[0] = 0;

  SServerStatusRsp statusRsp = {0};
  SMonMloadInfo    minfo = {0};
S
Shengliang Guan 已提交
159
  (*pMgmt->getMnodeLoadsFp)(&minfo);
dengyihao's avatar
dengyihao 已提交
160 161
  if (minfo.isMnode &&
      (minfo.load.syncState == TAOS_SYNC_STATE_ERROR || minfo.load.syncState == TAOS_SYNC_STATE_OFFLINE)) {
162 163 164 165 166 167
    pStatus->statusCode = TSDB_SRV_STATUS_SERVICE_DEGRADED;
    snprintf(pStatus->details, sizeof(pStatus->details), "mnode sync state is %s", syncStr(minfo.load.syncState));
    return;
  }

  SMonVloadInfo vinfo = {0};
S
Shengliang Guan 已提交
168
  (*pMgmt->getVnodeLoadsFp)(&vinfo);
169 170
  for (int32_t i = 0; i < taosArrayGetSize(vinfo.pVloads); ++i) {
    SVnodeLoad *pLoad = taosArrayGet(vinfo.pVloads, i);
S
Shengliang Guan 已提交
171
    if (pLoad->syncState == TAOS_SYNC_STATE_ERROR || pLoad->syncState == TAOS_SYNC_STATE_OFFLINE) {
172 173 174 175 176 177 178 179 180 181
      pStatus->statusCode = TSDB_SRV_STATUS_SERVICE_DEGRADED;
      snprintf(pStatus->details, sizeof(pStatus->details), "vnode:%d sync state is %s", pLoad->vgId,
               syncStr(pLoad->syncState));
      break;
    }
  }

  taosArrayDestroy(vinfo.pVloads);
}

S
Shengliang Guan 已提交
182
int32_t dmProcessServerRunStatus(SDnodeMgmt *pMgmt, SRpcMsg *pMsg) {
183 184 185 186
  dDebug("server run status req is received");
  SServerStatusRsp statusRsp = {0};
  dmGetServerRunStatus(pMgmt, &statusRsp);

187
  SRpcMsg rspMsg = {.info = pMsg->info};
188 189 190 191 192 193 194 195 196 197 198 199 200
  int32_t rspLen = tSerializeSServerStatusRsp(NULL, 0, &statusRsp);
  if (rspLen < 0) {
    rspMsg.code = TSDB_CODE_OUT_OF_MEMORY;
    return -1;
  }

  void *pRsp = rpcMallocCont(rspLen);
  if (pRsp == NULL) {
    rspMsg.code = TSDB_CODE_OUT_OF_MEMORY;
    return -1;
  }

  tSerializeSServerStatusRsp(pRsp, rspLen, &statusRsp);
S
Shengliang Guan 已提交
201 202
  pMsg->info.rsp = pRsp;
  pMsg->info.rspLen = rspLen;
203 204 205
  return 0;
}

206 207
SSDataBlock *dmBuildVariablesBlock(void) {
  SSDataBlock         *pBlock = taosMemoryCalloc(1, sizeof(SSDataBlock));
D
dapan1121 已提交
208
  size_t               size = 0;
209
  const SSysTableMeta *pMeta = NULL;
D
dapan1121 已提交
210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234
  getInfosDbMeta(&pMeta, &size);

  int32_t index = 0;
  for (int32_t i = 0; i < size; ++i) {
    if (strcmp(pMeta[i].name, TSDB_INS_TABLE_DNODE_VARIABLES) == 0) {
      index = i;
      break;
    }
  }

  pBlock->pDataBlock = taosArrayInit(pMeta[index].colNum, sizeof(SColumnInfoData));

  for (int32_t i = 0; i < pMeta[index].colNum; ++i) {
    SColumnInfoData colInfoData = {0};
    colInfoData.info.colId = i + 1;
    colInfoData.info.type = pMeta[index].schema[i].type;
    colInfoData.info.bytes = pMeta[index].schema[i].bytes;
    taosArrayPush(pBlock->pDataBlock, &colInfoData);
  }

  pBlock->info.hasVarCol = true;

  return pBlock;
}

235
int32_t dmAppendVariablesToBlock(SSDataBlock *pBlock, int32_t dnodeId) {
D
dapan1121 已提交
236 237 238 239 240 241
  int32_t numOfCfg = taosArrayGetSize(tsCfg->array);
  int32_t numOfRows = 0;
  blockDataEnsureCapacity(pBlock, numOfCfg);

  for (int32_t i = 0, c = 0; i < numOfCfg; ++i, c = 0) {
    SConfigItem *pItem = taosArrayGet(tsCfg->array, i);
wafwerar's avatar
wafwerar 已提交
242
    GRANT_CFG_SKIP;
D
dapan1121 已提交
243 244

    SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, c++);
245
    colDataSetVal(pColInfo, i, (const char *)&dnodeId, false);
D
dapan1121 已提交
246 247 248 249

    char name[TSDB_CONFIG_OPTION_LEN + VARSTR_HEADER_SIZE] = {0};
    STR_WITH_MAXSIZE_TO_VARSTR(name, pItem->name, TSDB_CONFIG_OPTION_LEN + VARSTR_HEADER_SIZE);
    pColInfo = taosArrayGet(pBlock->pDataBlock, c++);
250
    colDataSetVal(pColInfo, i, name, false);
251 252

    char    value[TSDB_CONFIG_VALUE_LEN + VARSTR_HEADER_SIZE] = {0};
D
dapan1121 已提交
253 254 255 256
    int32_t valueLen = 0;
    cfgDumpItemValue(pItem, &value[VARSTR_HEADER_SIZE], TSDB_CONFIG_VALUE_LEN, &valueLen);
    varDataSetLen(value, valueLen);
    pColInfo = taosArrayGet(pBlock->pDataBlock, c++);
257
    colDataSetVal(pColInfo, i, value, false);
D
dapan1121 已提交
258 259 260 261 262

    numOfRows++;
  }

  pBlock->info.rows = numOfRows;
263

D
dapan1121 已提交
264 265 266 267
  return TSDB_CODE_SUCCESS;
}

int32_t dmProcessRetrieve(SDnodeMgmt *pMgmt, SRpcMsg *pMsg) {
268 269
  int32_t size = 0;
  int32_t rowsRead = 0;
270

D
dapan1121 已提交
271 272 273 274 275 276
  SRetrieveTableReq retrieveReq = {0};
  if (tDeserializeSRetrieveTableReq(pMsg->pCont, pMsg->contLen, &retrieveReq) != 0) {
    terrno = TSDB_CODE_INVALID_MSG;
    return -1;
  }

277 278 279 280 281
  if (strcmp(retrieveReq.user, TSDB_DEFAULT_USER) != 0) {
    terrno = TSDB_CODE_MND_NO_RIGHTS;
    return -1;
  }

D
dapan1121 已提交
282 283 284 285 286
  if (strcasecmp(retrieveReq.tb, TSDB_INS_TABLE_DNODE_VARIABLES)) {
    terrno = TSDB_CODE_INVALID_MSG;
    return -1;
  }

287
  SSDataBlock *pBlock = dmBuildVariablesBlock();
D
dapan1121 已提交
288 289 290

  dmAppendVariablesToBlock(pBlock, pMgmt->pData->dnodeId);

S
shenglian zhou 已提交
291 292 293
  size_t numOfCols = taosArrayGetSize(pBlock->pDataBlock);
  size = sizeof(SRetrieveMetaTableRsp) + sizeof(int32_t) + sizeof(SSysTableSchema) * numOfCols +
         blockDataGetSize(pBlock) + blockDataGetSerialMetaSize(numOfCols);
D
dapan1121 已提交
294 295 296 297 298 299 300 301 302

  SRetrieveMetaTableRsp *pRsp = rpcMallocCont(size);
  if (pRsp == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    dError("failed to retrieve data since %s", terrstr());
    blockDataDestroy(pBlock);
    return -1;
  }

303
  char *pStart = pRsp->data;
S
shenglian zhou 已提交
304
  *(int32_t *)pStart = htonl(numOfCols);
D
dapan1121 已提交
305 306
  pStart += sizeof(int32_t);  // number of columns

S
shenglian zhou 已提交
307
  for (int32_t i = 0; i < numOfCols; ++i) {
D
dapan1121 已提交
308 309
    SSysTableSchema *pSchema = (SSysTableSchema *)pStart;
    SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, i);
310

D
dapan1121 已提交
311 312 313 314 315 316 317
    pSchema->bytes = htonl(pColInfo->info.bytes);
    pSchema->colId = htons(pColInfo->info.colId);
    pSchema->type = pColInfo->info.type;

    pStart += sizeof(SSysTableSchema);
  }

H
Haojun Liao 已提交
318
  int32_t len = blockEncode(pBlock, pStart, numOfCols);
D
dapan1121 已提交
319 320 321 322 323 324 325 326 327 328 329 330

  pRsp->numOfRows = htonl(pBlock->info.rows);
  pRsp->precision = TSDB_TIME_PRECISION_MILLI;  // millisecond time precision
  pRsp->completed = 1;
  pMsg->info.rsp = pRsp;
  pMsg->info.rspLen = size;
  dDebug("dnode variables retrieve completed");

  blockDataDestroy(pBlock);
  return TSDB_CODE_SUCCESS;
}

S
Shengliang Guan 已提交
331 332 333 334 335 336 337 338 339 340 341 342 343
SArray *dmGetMsgHandles() {
  int32_t code = -1;
  SArray *pArray = taosArrayInit(16, sizeof(SMgmtHandle));
  if (pArray == NULL) goto _OVER;

  // Requests handled by DNODE
  if (dmSetMgmtHandle(pArray, TDMT_DND_CREATE_MNODE, dmPutNodeMsgToMgmtQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_DND_DROP_MNODE, dmPutNodeMsgToMgmtQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_DND_CREATE_QNODE, dmPutNodeMsgToMgmtQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_DND_DROP_QNODE, dmPutNodeMsgToMgmtQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_DND_CREATE_SNODE, dmPutNodeMsgToMgmtQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_DND_DROP_SNODE, dmPutNodeMsgToMgmtQueue, 0) == NULL) goto _OVER;
  if (dmSetMgmtHandle(pArray, TDMT_DND_CONFIG_DNODE, dmPutNodeMsgToMgmtQueue, 0) == NULL) goto _OVER;
344
  if (dmSetMgmtHandle(pArray, TDMT_DND_SERVER_STATUS, dmPutNodeMsgToMgmtQueue, 0) == NULL) goto _OVER;
D
dapan1121 已提交
345
  if (dmSetMgmtHandle(pArray, TDMT_DND_SYSTABLE_RETRIEVE, dmPutNodeMsgToMgmtQueue, 0) == NULL) goto _OVER;
S
Shengliang Guan 已提交
346 347

  // Requests handled by MNODE
C
Cary Xu 已提交
348
  if (dmSetMgmtHandle(pArray, TDMT_MND_GRANT, dmPutNodeMsgToMgmtQueue, 0) == NULL) goto _OVER;
S
Shengliang Guan 已提交
349 350 351 352 353 354 355 356 357 358 359 360
  if (dmSetMgmtHandle(pArray, TDMT_MND_AUTH_RSP, dmPutNodeMsgToMgmtQueue, 0) == NULL) goto _OVER;

  code = 0;

_OVER:
  if (code != 0) {
    taosArrayDestroy(pArray);
    return NULL;
  } else {
    return pArray;
  }
}