dnodeDnode.c

/*
 * 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 "dnodeDnode.h"
#include "dnodeTransport.h"
#include "tthread.h"
#include "ttime.h"
#include "cJSON.h"
#include "thash.h"

static struct {
  int32_t         dnodeId;
  int32_t         dropped;
  int64_t         clusterId;
  SDnodeEps      *dnodeEps;
  SHashObj       *dnodeHash;
  SRpcEpSet       mnodeEpSetForShell;
  SRpcEpSet       mnodeEpSetForPeer;
  char            file[PATH_MAX + 20];
  pthread_mutex_t mutex;
} tsConfig;

void dnodeGetEpSetForPeer(SRpcEpSet *epSet) {
  pthread_mutex_lock(&tsConfig.mutex);
  *epSet = tsConfig.mnodeEpSetForPeer;
  pthread_mutex_unlock(&tsConfig.mutex);
}

static void dnodeGetEpSetForShell(SRpcEpSet *epSet) {
  pthread_mutex_lock(&tsConfig.mutex);
  *epSet = tsConfig.mnodeEpSetForShell;
  pthread_mutex_unlock(&tsConfig.mutex);
}

void dnodeUpdateMnodeEps(SRpcEpSet *ep) {
  if (ep != NULL || ep->numOfEps <= 0) {
    dError("mnode is changed, but content is invalid, discard it");
    return;
  }

  pthread_mutex_lock(&tsConfig.mutex);

  dInfo("mnode is changed, num:%d use:%d", ep->numOfEps, ep->inUse);

  tsConfig.mnodeEpSetForPeer = *ep;
  for (int32_t i = 0; i < ep->numOfEps; ++i) {
    ep->port[i] -= TSDB_PORT_DNODEDNODE;
    dInfo("mnode index:%d %s:%u", i, ep->fqdn[i], ep->port[i]);
  }
  tsConfig.mnodeEpSetForShell = *ep;

  pthread_mutex_unlock(&tsConfig.mutex);
}

void dnodeSendRedirectMsg(SRpcMsg *rpcMsg, bool forShell) {
  SRpcConnInfo connInfo = {0};
  rpcGetConnInfo(rpcMsg->handle, &connInfo);

  SRpcEpSet epSet = {0};
  if (forShell) {
    dnodeGetEpSetForShell(&epSet);
  } else {
    dnodeGetEpSetForPeer(&epSet);
  }

  dDebug("msg:%s will be redirected, num:%d use:%d", taosMsg[rpcMsg->msgType], epSet.numOfEps, epSet.inUse);

  for (int32_t i = 0; i < epSet.numOfEps; ++i) {
    dDebug("mnode index:%d %s:%d", i, epSet.fqdn[i], epSet.port[i]);
    if (strcmp(epSet.fqdn[i], tsLocalFqdn) == 0) {
      if ((epSet.port[i] == tsServerPort + TSDB_PORT_DNODEDNODE && !forShell) ||
          (epSet.port[i] == tsServerPort && forShell)) {
        epSet.inUse = (i + 1) % epSet.numOfEps;
        dDebug("mnode index:%d %s:%d set inUse to %d", i, epSet.fqdn[i], epSet.port[i], epSet.inUse);
      }
    }

    epSet.port[i] = htons(epSet.port[i]);
  }

  rpcSendRedirectRsp(rpcMsg->handle, &epSet);
}

static void dnodePrintEps() {
  dDebug("print dnode list, num:%d", tsConfig.dnodeEps->dnodeNum);
  for (int32_t i = 0; i < tsConfig.dnodeEps->dnodeNum; i++) {
    SDnodeEp *ep = &tsConfig.dnodeEps->dnodeEps[i];
    dDebug("dnode:%d, fqdn:%s port:%u isMnode:%d", ep->dnodeId, ep->dnodeFqdn, ep->dnodePort, ep->isMnode);
  }
}

static void dnodeResetEps(SDnodeEps *data) {
  assert(data != NULL);

  int32_t size = sizeof(SDnodeEps) + data->dnodeNum * sizeof(SDnodeEp);

  if (data->dnodeNum > tsConfig.dnodeEps->dnodeNum) {
    SDnodeEps *tmp = calloc(1, size);
    if (tmp == NULL) return;

    tfree(tsConfig.dnodeEps);
    tsConfig.dnodeEps = tmp;
  }

  if (tsConfig.dnodeEps != data) {
    memcpy(tsConfig.dnodeEps, data, size);
  }

  tsConfig.mnodeEpSetForPeer.inUse = 0;
  tsConfig.mnodeEpSetForShell.inUse = 0;
  int32_t index = 0;
  for (int32_t i = 0; i < tsConfig.dnodeEps->dnodeNum; i++) {
    SDnodeEp *ep = &tsConfig.dnodeEps->dnodeEps[i];
    if (!ep->isMnode) continue;
    if (index >= TSDB_MAX_REPLICA) continue;
    strcpy(tsConfig.mnodeEpSetForShell.fqdn[index], ep->dnodeFqdn);
    strcpy(tsConfig.mnodeEpSetForPeer.fqdn[index], ep->dnodeFqdn);
    tsConfig.mnodeEpSetForShell.port[index] = ep->dnodePort;
    tsConfig.mnodeEpSetForShell.port[index] = ep->dnodePort + tsDnodeDnodePort;
    index++;
  }

  for (int32_t i = 0; i < tsConfig.dnodeEps->dnodeNum; ++i) {
    SDnodeEp *ep = &tsConfig.dnodeEps->dnodeEps[i];
    taosHashPut(tsConfig.dnodeHash, &ep->dnodeId, sizeof(int32_t), ep, sizeof(SDnodeEp));
  }

  dnodePrintEps();
}

static bool dnodeIsDnodeEpChanged(int32_t dnodeId, char *epstr) {
  bool changed = false;

  pthread_mutex_lock(&tsConfig.mutex);

  SDnodeEp *ep = taosHashGet(tsConfig.dnodeHash, &dnodeId, sizeof(int32_t));
  if (ep != NULL) {
    char epSaved[TSDB_EP_LEN + 1];
    snprintf(epSaved, TSDB_EP_LEN, "%s:%u", ep->dnodeFqdn, ep->dnodePort);
    changed = strcmp(epstr, epSaved) != 0;
    tstrncpy(epstr, epSaved, TSDB_EP_LEN);
  }

  pthread_mutex_unlock(&tsConfig.mutex);

  return changed;
}

static int32_t dnodeReadEps() {
  int32_t len = 0;
  int32_t maxLen = 30000;
  char   *content = calloc(1, maxLen + 1);
  cJSON  *root = NULL;
  FILE   *fp = NULL;

  fp = fopen(tsConfig.file, "r");
  if (!fp) {
    dDebug("file %s not exist", tsConfig.file);
    goto PRASE_EPS_OVER;
  }

  len = (int32_t)fread(content, 1, maxLen, fp);
  if (len <= 0) {
    dError("failed to read %s since content is null", tsConfig.file);
    goto PRASE_EPS_OVER;
  }

  content[len] = 0;
  root = cJSON_Parse(content);
  if (root == NULL) {
    dError("failed to read %s since invalid json format", tsConfig.file);
    goto PRASE_EPS_OVER;
  }

  cJSON *dnodeId = cJSON_GetObjectItem(root, "dnodeId");
  if (!dnodeId || dnodeId->type != cJSON_String) {
    dError("failed to read %s since dnodeId not found", tsConfig.file);
    goto PRASE_EPS_OVER;
  }
  tsConfig.dnodeId = atoi(dnodeId->valuestring);

  cJSON *dropped = cJSON_GetObjectItem(root, "dropped");
  if (!dropped || dropped->type != cJSON_String) {
    dError("failed to read %s since dropped not found", tsConfig.file);
    goto PRASE_EPS_OVER;
  }
  tsConfig.dropped = atoi(dropped->valuestring);

  cJSON *clusterId = cJSON_GetObjectItem(root, "clusterId");
  if (!clusterId || clusterId->type != cJSON_String) {
    dError("failed to read %s since clusterId not found", tsConfig.file);
    goto PRASE_EPS_OVER;
  }
  tsConfig.clusterId = atoll(clusterId->valuestring);

  cJSON *dnodeInfos = cJSON_GetObjectItem(root, "dnodeInfos");
  if (!dnodeInfos || dnodeInfos->type != cJSON_Array) {
    dError("failed to read %s since dnodeInfos not found", tsConfig.file);
    goto PRASE_EPS_OVER;
  }

  int32_t dnodeInfosSize = cJSON_GetArraySize(dnodeInfos);
  if (dnodeInfosSize <= 0) {
    dError("failed to read %s since dnodeInfos size:%d invalid", tsConfig.file, dnodeInfosSize);
    goto PRASE_EPS_OVER;
  }

  tsConfig.dnodeEps = calloc(1, dnodeInfosSize * sizeof(SDnodeEp) + sizeof(SDnodeEps));
  if (tsConfig.dnodeEps == NULL) {
    dError("failed to calloc dnodeEpList since %s", strerror(errno));
    goto PRASE_EPS_OVER;
  }
  tsConfig.dnodeEps->dnodeNum = dnodeInfosSize;

  for (int32_t i = 0; i < dnodeInfosSize; ++i) {
    cJSON *dnodeInfo = cJSON_GetArrayItem(dnodeInfos, i);
    if (dnodeInfo == NULL) break;

    SDnodeEp *ep = &tsConfig.dnodeEps->dnodeEps[i];

    cJSON *dnodeId = cJSON_GetObjectItem(dnodeInfo, "dnodeId");
    if (!dnodeId || dnodeId->type != cJSON_String) {
      dError("failed to read %s, dnodeId not found", tsConfig.file);
      goto PRASE_EPS_OVER;
    }
    ep->dnodeId = atoi(dnodeId->valuestring);

    cJSON *isMnode = cJSON_GetObjectItem(dnodeInfo, "isMnode");
    if (!isMnode || isMnode->type != cJSON_String) {
      dError("failed to read %s, isMnode not found", tsConfig.file);
      goto PRASE_EPS_OVER;
    }
    ep->isMnode = atoi(isMnode->valuestring);

    cJSON *dnodeFqdn = cJSON_GetObjectItem(dnodeInfo, "dnodeFqdn");
    if (!dnodeFqdn || dnodeFqdn->type != cJSON_String || dnodeFqdn->valuestring == NULL) {
      dError("failed to read %s, dnodeFqdn not found", tsConfig.file);
      goto PRASE_EPS_OVER;
    }
    tstrncpy(ep->dnodeFqdn, dnodeFqdn->valuestring, TSDB_FQDN_LEN);

    cJSON *dnodePort = cJSON_GetObjectItem(dnodeInfo, "dnodePort");
    if (!dnodePort || dnodePort->type != cJSON_String) {
      dError("failed to read %s, dnodePort not found", tsConfig.file);
      goto PRASE_EPS_OVER;
    }
    ep->dnodePort = atoi(dnodePort->valuestring);
  }

  dInfo("succcessed to read file %s", tsConfig.file);
  dnodePrintEps();

PRASE_EPS_OVER:
  if (content != NULL) free(content);
  if (root != NULL) cJSON_Delete(root);
  if (fp != NULL) fclose(fp);

  if (dnodeIsDnodeEpChanged(tsConfig.dnodeId, tsLocalEp)) {
    dError("dnode:%d, localEp %s different with dnodeEps.json and need reconfigured", tsConfig.dnodeId, tsLocalEp);
    return -1;
  }

  dnodeResetEps(tsConfig.dnodeEps);

  terrno = 0;
  return 0;
}

static int32_t dnodeWriteEps() {
  FILE *fp = fopen(tsConfig.file, "w");
  if (!fp) {
    dError("failed to write %s since %s", tsConfig.file, strerror(errno));
    return -1;
  }

  int32_t len = 0;
  int32_t maxLen = 30000;
  char   *content = calloc(1, maxLen + 1);

  len += snprintf(content + len, maxLen - len, "{\n");
  len += snprintf(content + len, maxLen - len, "  \"dnodeId\": \"%d\",\n", tsConfig.dnodeId);
  len += snprintf(content + len, maxLen - len, "  \"dropped\": \"%d\",\n", tsConfig.dropped);
  len += snprintf(content + len, maxLen - len, "  \"clusterId\": \"%" PRId64 "\",\n", tsConfig.clusterId);
  len += snprintf(content + len, maxLen - len, "  \"dnodeInfos\": [{\n");
  for (int32_t i = 0; i < tsConfig.dnodeEps->dnodeNum; ++i) {
    SDnodeEp *ep = &tsConfig.dnodeEps->dnodeEps[i];
    len += snprintf(content + len, maxLen - len, "    \"dnodeId\": \"%d\",\n", ep->dnodeId);
    len += snprintf(content + len, maxLen - len, "    \"isMnode\": \"%d\",\n", ep->isMnode);
    len += snprintf(content + len, maxLen - len, "    \"dnodeFqdn\": \"%s\",\n", ep->dnodeFqdn);
    len += snprintf(content + len, maxLen - len, "    \"dnodePort\": \"%u\"\n", ep->dnodePort);
    if (i < tsConfig.dnodeEps->dnodeNum - 1) {
      len += snprintf(content + len, maxLen - len, "  },{\n");
    } else {
      len += snprintf(content + len, maxLen - len, "  }]\n");
    }
  }
  len += snprintf(content + len, maxLen - len, "}\n");

  fwrite(content, 1, len, fp);
  taosFsyncFile(fileno(fp));
  fclose(fp);
  free(content);
  terrno = 0;

  dInfo("successed to write %s", tsConfig.file);
  return 0;
}

int32_t dnodeInitConfig() {
  tsConfig.dnodeId = 0;
  tsConfig.dropped = 0;
  tsConfig.clusterId = 0;
  tsConfig.dnodeEps = NULL;
  snprintf(tsConfig.file, sizeof(tsConfig.file), "%s/dnodeEps.json", tsDnodeDir);
  pthread_mutex_init(&tsConfig.mutex, NULL);

  tsConfig.dnodeHash = taosHashInit(4, taosGetDefaultHashFunction(TSDB_DATA_TYPE_INT), true, HASH_ENTRY_LOCK);
  if (tsConfig.dnodeHash == NULL) return -1;

  int32_t ret = dnodeReadEps();
  if (ret == 0) {
    dInfo("dnode eps is initialized");
  }

  return ret;
}

void dnodeCleanupConfig() {
  pthread_mutex_lock(&tsConfig.mutex);

  if (tsConfig.dnodeEps != NULL) {
    free(tsConfig.dnodeEps);
    tsConfig.dnodeEps = NULL;
  }

  if (tsConfig.dnodeHash) {
    taosHashCleanup(tsConfig.dnodeHash);
    tsConfig.dnodeHash = NULL;
  }

  pthread_mutex_unlock(&tsConfig.mutex);
  pthread_mutex_destroy(&tsConfig.mutex);
}

void dnodeUpdateDnodeEps(SDnodeEps *data) {
  if (data == NULL || data->dnodeNum <= 0) return;

  pthread_mutex_lock(&tsConfig.mutex);

  if (data->dnodeNum != tsConfig.dnodeEps->dnodeNum) {
    dnodeResetEps(data);
    dnodeWriteEps();
  } else {
    int32_t size = data->dnodeNum * sizeof(SDnodeEp) + sizeof(SDnodeEps);
    if (memcmp(tsConfig.dnodeEps, data, size) != 0) {
      dnodeResetEps(data);
      dnodeWriteEps();
    }
  }

  pthread_mutex_unlock(&tsConfig.mutex);
}

void dnodeGetEp(int32_t dnodeId, char *epstr, char *fqdn, uint16_t *port) {
  pthread_mutex_lock(&tsConfig.mutex);

  SDnodeEp *ep = taosHashGet(tsConfig.dnodeHash, &dnodeId, sizeof(int32_t));
  if (ep != NULL) {
    if (port) *port = ep->dnodePort;
    if (fqdn) tstrncpy(fqdn, ep->dnodeFqdn, TSDB_FQDN_LEN);
    if (epstr) snprintf(epstr, TSDB_EP_LEN, "%s:%u", ep->dnodeFqdn, ep->dnodePort);
  }

  pthread_mutex_unlock(&tsConfig.mutex);
}

void dnodeUpdateCfg(SDnodeCfg *data) {
  if (tsConfig.dnodeId != 0 && !data->dropped) return;

  pthread_mutex_lock(&tsConfig.mutex);

  tsConfig.dnodeId = data->dnodeId;
  tsConfig.clusterId = data->clusterId;
  tsConfig.dropped = data->dropped;
  dInfo("dnodeId is set to %d, clusterId is set to %" PRId64, data->dnodeId, data->clusterId);

  dnodeWriteEps();
  pthread_mutex_unlock(&tsConfig.mutex);
}

int32_t dnodeGetDnodeId() {
  int32_t dnodeId = 0;
  pthread_mutex_lock(&tsConfig.mutex);
  dnodeId = tsConfig.dnodeId;
  pthread_mutex_unlock(&tsConfig.mutex);
  return dnodeId;
}

int64_t dnodeGetClusterId() {
  int64_t clusterId = 0;
  pthread_mutex_lock(&tsConfig.mutex);
  clusterId = tsConfig.clusterId;
  pthread_mutex_unlock(&tsConfig.mutex);
  return clusterId;
}

static struct {
  pthread_t *threadId;
  bool       threadStop;
  uint32_t   rebootTime;
} tsDnode;

static void dnodeSendStatusMsg() {
  int32_t     contLen = sizeof(SStatusMsg) + TSDB_MAX_VNODES * sizeof(SVnodeLoad);
  SStatusMsg *pStatus = rpcMallocCont(contLen);
  if (pStatus == NULL) {
    dError("failed to malloc status message");
    return;
  }

  pStatus->version = htonl(tsVersion);
  pStatus->dnodeId = htonl(dnodeGetDnodeId());
  tstrncpy(pStatus->dnodeEp, tsLocalEp, TSDB_EP_LEN);
  pStatus->clusterId = htobe64(dnodeGetClusterId());
  pStatus->lastReboot = htonl(tsDnode.rebootTime);
  pStatus->numOfCores = htonl(tsNumOfCores);
  pStatus->diskAvailable = tsAvailDataDirGB;

  // fill cluster cfg parameters
  pStatus->clusterCfg.statusInterval = htonl(tsStatusInterval);
  pStatus->clusterCfg.checkTime = 0;
  tstrncpy(pStatus->clusterCfg.timezone, tsTimezone, 64);
  char timestr[32] = "1970-01-01 00:00:00.00";
  (void)taosParseTime(timestr, &pStatus->clusterCfg.checkTime, (int32_t)strlen(timestr), TSDB_TIME_PRECISION_MILLI, 0);
  tstrncpy(pStatus->clusterCfg.locale, tsLocale, TSDB_LOCALE_LEN);
  tstrncpy(pStatus->clusterCfg.charset, tsCharset, TSDB_LOCALE_LEN);

  // vnodeGetStatus(NULL, pStatus);
  // contLen = sizeof(SStatusMsg) + pStatus->openVnodes * sizeof(SVnodeLoad);
  // pStatus->openVnodes = htons(pStatus->openVnodes);

  SRpcMsg rpcMsg = {.ahandle = NULL, .pCont = pStatus, .contLen = contLen, .msgType = TSDB_MSG_TYPE_DM_STATUS};

  dnodeSendMsgToMnode(&rpcMsg);
}

void dnodeProcessStatusRsp(SRpcMsg *pMsg) {
  dTrace("status rsp is received, code:%s", tstrerror(pMsg->code));
  if (pMsg->code != TSDB_CODE_SUCCESS) return;

  SStatusRsp *pStatusRsp = pMsg->pCont;

  SDnodeCfg *pCfg = &pStatusRsp->dnodeCfg;
  pCfg->dnodeId = htonl(pCfg->dnodeId);
  pCfg->clusterId = htobe64(pCfg->clusterId);
  pCfg->numOfVnodes = htonl(pCfg->numOfVnodes);
  pCfg->numOfDnodes = htonl(pCfg->numOfDnodes);
  dnodeUpdateCfg(pCfg);

  if (pCfg->dropped) {
    dError("status rsp is received, and set dnode to drop status");
    return;
  }

  // vnodeSetAccess(pStatusRsp->vgAccess, pCfg->numOfVnodes);

  SDnodeEps *eps = (SDnodeEps *)((char *)pStatusRsp->vgAccess + pCfg->numOfVnodes * sizeof(SVgroupAccess));
  eps->dnodeNum = htonl(eps->dnodeNum);
  for (int32_t i = 0; i < eps->dnodeNum; ++i) {
    eps->dnodeEps[i].dnodeId = htonl(eps->dnodeEps[i].dnodeId);
    eps->dnodeEps[i].dnodePort = htons(eps->dnodeEps[i].dnodePort);
  }

  dnodeUpdateDnodeEps(eps);
}

static void *dnodeThreadRoutine(void *param) {
  int32_t ms = tsStatusInterval * 1000;
  while (!tsDnode.threadStop) {
    taosMsleep(ms);
    dnodeSendStatusMsg();
  }
}

int32_t dnodeInitDnode() {
  tsDnode.threadStop = false;
  tsDnode.rebootTime = taosGetTimestampSec();
  tsDnode.threadId = taosCreateThread(dnodeThreadRoutine, NULL);
  if (tsDnode.threadId == NULL) {
    return -1;
  }

  dInfo("dnode msg is initialized");
  return 0;
}

void dnodeCleanupDnode() {
  if (tsDnode.threadId != NULL) {
    tsDnode.threadStop = true;
    taosDestoryThread(tsDnode.threadId);
    tsDnode.threadId = NULL;
  }

  dInfo("dnode msg is cleanuped");
}

void dnodeProcessConfigDnodeReq(SRpcMsg *pMsg) {
  SCfgDnodeMsg *pCfg = pMsg->pCont;

  int32_t code = taosCfgDynamicOptions(pCfg->config);

  SRpcMsg rspMsg = {.handle = pMsg->handle, .pCont = NULL, .contLen = 0, .code = code};

  rpcSendResponse(&rspMsg);
  rpcFreeCont(pMsg->pCont);
}

void dnodeProcessStartupReq(SRpcMsg *pMsg) {
  dInfo("startup msg is received, cont:%s", (char *)pMsg->pCont);

  SStartupStep *pStep = rpcMallocCont(sizeof(SStartupStep));
  dnodeGetStartup(pStep);

  dDebug("startup msg is sent, step:%s desc:%s finished:%d", pStep->name, pStep->desc, pStep->finished);

  SRpcMsg rpcRsp = {.handle = pMsg->handle, .pCont = pStep, .contLen = sizeof(SStartupStep)};
  rpcSendResponse(&rpcRsp);
  rpcFreeCont(pMsg->pCont);
}