#3834

Committed 03 Apr 2025 10:04AM UTC coverage: 33.925% (-0.005%) from 33.93%

Build # #3834

Build Type

push

travis-ci

Committed by GitHub

Commit Message

merge: from main to 3.0 branch

Coverage Stats

149101 of 604631 branches covered (24.66%)

Branch coverage included in aggregate %.

223887 of 494810 relevant lines covered (45.25%)

760082.74 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

47.9

/enterprise/src/plugins/dnode/src/dmodule.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 "dmMgmt.h"
#include "tchecksum.h"

#if defined(GRANTS_CFG) || defined(_TD_DARWIN_64)
#define _TD_DM_SKIP_CHECK
#endif

#if !defined(ASSERT_NOT_CORE) && !defined(WINDOWS)
#define _TD_DM_CHECK_OFFSET
#define DM_CHECK_OFFSET(p1, p2, offset, flag)             \
  do {                                                    \
    int32_t off = POINTER_DISTANCE((p1), (p2));           \
    if ((offset) != abs(off)) {                           \
      dError("%s offset: %d!=%d", (flag), off, (offset)); \
      return TSDB_CODE_INTERNAL_ERROR;                    \
    }                                                     \
  } while (0)
#endif

extern char tsVersionName[];

typedef enum {
  DM_ENG_FVER_1 = 1,
} DM_ENG_FVER;

typedef enum {
  DM_ETYPE_UN = 0,
  DM_ETYPE_OS = 1,
  DM_ETYPE_TR = 2,
  DM_ETYPE_EN = 3,
} DM_ENG_TYPE;

#define DM_ENG_FVER_MAX DM_ENG_FVER_1
#define DM_OS_ST_NAME_LEN 64

#define DM_ENGINE_FILE "dnode.info"
#define DM_ENGINE_FILE_T "dnode.info.t"
#define DNODE_CFG_FILE "dnode.json"

typedef struct {
  int8_t  type;  // 0 unknown 1 community 2 trial 3 official
  int32_t dnodeId;
  int32_t engineVer;  // tsVersion
  int64_t clusterId;
  int64_t createMs;
  int64_t updateMs;
} SEngineInfo;

// file operation (TODO: refactor to dmUtil.c)
#define DM_FILE_HEAD_SIZE 512
typedef struct {
  uint32_t version;
  uint32_t len;  // Encoded content len(checksum included)
} SDFHeader;

#define STR_CASE_CMP(s, d) (0 == strcasecmp((s), (d)))
#define STR_STR_CMP(s, d) (strstr((s), (d)))
#define STR_INT_CMP(s, d, c) (taosStr2Int32(s, 0, 10) c(d))
#define STR_STR_SIGN ("ia")
#define STR_STR_COMM ("unit")

#define STR_CASE_STR_CHECK(s, d)   \
  do {                             \
    TAOS_UNUSED(strtolower(s, s)); \
    TAOS_UNUSED(strtolower(d, d)); \
    if (STR_STR_CMP(s, d)) {       \
      DM_ERR_RTN(0);               \
    }                              \
  } while (0)

#define DM_ERR_RTN(c) \
  do {                \
    code = (c);       \
    goto _exit;       \
  } while (0)

static const char *dmOS[] = {"Ubuntu", "CentOS Linux", "Red Hat", "Debian GNU/Linux", "CoreOS", "FreeBSD", "openSUSE",
                             "SLES",   "Fedora",       "macOS",   "CentOS Stream"};

// declarations
static void    dmFetchEType(int8_t *type);
static void    dmGetFname(const char *fname, char *ofname);
static int32_t dmSyncEps(SDnodeData *pData);
static int32_t dmEncodeVars(void *buf, int32_t bufLen, SEngineInfo *pInfo);
static int32_t dmEncodeVars(void *buf, int32_t bufLen, SEngineInfo *pInfo);
static int32_t dmReadVars(SEngineInfo *pInfo);
static int32_t dmWriteVars(SEngineInfo *pInfo);

// implementations

#ifdef _TD_DM_CHECK_OFFSET
static int32_t dmCheckOffset(SDnode *pDnode) {
  SDnodeData     *pData = &pDnode->data;
  TdThreadRwlock *pLock = &pData->lock;
  int32_t        *pDnodeId = &pData->dnodeId;
  int32_t        *pEngineVer = &pData->engineVer;
  int64_t        *pClusterId = &pData->clusterId;

  DM_CHECK_OFFSET(pData, pDnode, 168, "dnode data");
  DM_CHECK_OFFSET(pLock, pData, 720, "data lock");
  DM_CHECK_OFFSET(pDnodeId, pData, 0, "data dnodeId");
  DM_CHECK_OFFSET(pEngineVer, pData, 4, "data engineVer");
  DM_CHECK_OFFSET(pClusterId, pData, 8, "data clusterId");

  return TSDB_CODE_SUCCESS;
}
#endif

static int32_t dmInitPrerequisites() {
#ifndef _TD_DM_SKIP_CHECK
  int32_t code = 0;

  char reName[64] = {0};
  char stName[64] = {0};
  char ver[64] = {0};

  code = (int32_t)(2147483648 | 298);
  tstrncpy(stName, tsVersionName, 16);

  if (STR_STR_CMP(stName, STR_STR_SIGN)) {
    DM_ERR_RTN(0);
  }
  if (taosGetOsReleaseName(reName, stName, ver, DM_OS_ST_NAME_LEN) != 0) {
    int32_t errCode = TAOS_SYSTEM_ERROR(errno);
    if (errCode != 0) code = errCode;
    TAOS_CHECK_GOTO(code, NULL, _exit);
  }
  if (STR_CASE_CMP(stName, dmOS[0])) {
    if (STR_INT_CMP(ver, 17, >)) {
      DM_ERR_RTN(0);
    }
  } else if (STR_CASE_CMP(stName, dmOS[1])) {
    if (STR_INT_CMP(ver, 6, >)) {
      DM_ERR_RTN(0);
    }
  } else {
    int32_t size = sizeof(dmOS) / sizeof(dmOS[0]);
    char    os[DM_OS_ST_NAME_LEN] = {0};
    for (int32_t i = 2; i < size; ++i) {
      tstrncpy(os, dmOS[i], DM_OS_ST_NAME_LEN);
      STR_CASE_STR_CHECK(stName, os);
    }
  }

_exit:
  TAOS_RETURN(code);
#else
  TAOS_RETURN(0);
#endif
}

static int dmEncodeDFHeader(void **buf, SDFHeader *pHeader) {
  int tlen = 0;

  tlen += taosEncodeFixedU32(buf, pHeader->version);
  tlen += taosEncodeFixedU32(buf, pHeader->len);

  return tlen;
}

static void *dmDecodeDFHeader(void *buf, SDFHeader *pHeader) {
  buf = taosDecodeFixedU32(buf, &(pHeader->version));
  buf = taosDecodeFixedU32(buf, &(pHeader->len));

  return buf;
}

static int32_t dmEncodeVars(void *buf, int32_t bufLen, SEngineInfo *pInfo) {
  SEncoder encoder = {0};
  tEncoderInit(&encoder, buf, bufLen);
  int32_t code = 0;

  TAOS_CHECK_GOTO(tStartEncode(&encoder), NULL, _exit);
  TAOS_CHECK_GOTO(tEncodeI8(&encoder, pInfo->type), NULL, _exit);
  TAOS_CHECK_GOTO(tEncodeI32v(&encoder, pInfo->dnodeId), NULL, _exit);
  TAOS_CHECK_GOTO(tEncodeI32v(&encoder, pInfo->engineVer), NULL, _exit);
  TAOS_CHECK_GOTO(tEncodeI64v(&encoder, pInfo->clusterId), NULL, _exit);
  TAOS_CHECK_GOTO(tEncodeI64v(&encoder, pInfo->createMs), NULL, _exit);
  TAOS_CHECK_GOTO(tEncodeI64v(&encoder, pInfo->updateMs), NULL, _exit);

  tEndEncode(&encoder);
  code = encoder.pos;
_exit:
  tEncoderClear(&encoder);
  return code;
}

static int32_t dmDecodeVars(void *buf, int32_t bufLen, SEngineInfo *pInfo) {
  SDecoder decoder = {0};
  tDecoderInit(&decoder, buf, bufLen);
  int32_t code = 0;

  TAOS_CHECK_GOTO(tStartDecode(&decoder), NULL, _exit);

  TAOS_CHECK_GOTO(tDecodeI8(&decoder, &pInfo->type), NULL, _exit);
  TAOS_CHECK_GOTO(tDecodeI32v(&decoder, &pInfo->dnodeId), NULL, _exit);
  TAOS_CHECK_GOTO(tDecodeI32v(&decoder, &pInfo->engineVer), NULL, _exit);
  TAOS_CHECK_GOTO(tDecodeI64v(&decoder, &pInfo->clusterId), NULL, _exit);
  TAOS_CHECK_GOTO(tDecodeI64v(&decoder, &pInfo->createMs), NULL, _exit);
  TAOS_CHECK_GOTO(tDecodeI64v(&decoder, &pInfo->updateMs), NULL, _exit);

  tEndDecode(&decoder);

_exit:
  tDecoderClear(&decoder);
  TAOS_RETURN(code);
}

static int32_t dmReadVars(SEngineInfo *pInfo) {
  int32_t   code = 0;
  int32_t   lino = 0;
  TdFilePtr pFile = NULL;
  void     *buffer = NULL;
  void     *ptr;
  SDFHeader dHeader;
  char      fname[FILENAME_MAX] = "\0";

  dmGetFname(DM_ENGINE_FILE, fname);

  errno = 0;  // clear errno

  if (!taosCheckExistFile(fname)) {
    TAOS_CHECK_GOTO(TSDB_CODE_NOT_FOUND, &lino, _exit);
  }

  pFile = taosOpenFile(fname, TD_FILE_READ);
  if (!pFile) {
    if (errno == ENOENT) {
      TAOS_CHECK_GOTO(TSDB_CODE_NOT_FOUND, &lino, _exit);
    } else {
      TAOS_CHECK_GOTO(TAOS_SYSTEM_ERROR(errno), &lino, _exit);
    }
  }

  if (!(buffer = taosMemoryMalloc(DM_FILE_HEAD_SIZE))) {
    TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, &lino, _exit);
  }

  int64_t nRead = taosReadFile(pFile, buffer, DM_FILE_HEAD_SIZE);
  if (nRead < 0) {
    TAOS_CHECK_GOTO(TAOS_SYSTEM_ERROR(errno), &lino, _exit);
  }

  if (nRead != DM_FILE_HEAD_SIZE) {
    code = TSDB_CODE_FILE_CORRUPTED;
    dTrace("failed to read %d bytes from vars head since %s", DM_FILE_HEAD_SIZE, tstrerror(code));
    TAOS_CHECK_GOTO(code, &lino, _exit);
  }

  if (!taosCheckChecksumWhole((uint8_t *)buffer, DM_FILE_HEAD_SIZE)) {
    dTrace("failed to read vars head since wrong checksum");
    TAOS_CHECK_GOTO(TSDB_CODE_CHECKSUM_ERROR, &lino, _exit);
  }

  ptr = buffer;
  ptr = dmDecodeDFHeader(ptr, &dHeader);

  if (dHeader.version != DM_ENG_FVER_1) {
    // TODO
  }

  if (dHeader.len > 0) {
    if (dHeader.len > DM_FILE_HEAD_SIZE) {
      void *tmpBuf = NULL;
      if (!(tmpBuf = taosMemoryRealloc(buffer, dHeader.len))) {
        TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, &lino, _exit);
      }
      buffer = tmpBuf;
    }

    nRead = (int)taosReadFile(pFile, buffer, dHeader.len);
    if (nRead < 0) {
      code = TAOS_SYSTEM_ERROR(errno);
      TAOS_CHECK_GOTO(code, &lino, _exit);
    }

    if (nRead != dHeader.len) {
      code = TSDB_CODE_FILE_CORRUPTED;
      dTrace("failed to read %d bytes from vars body since %s", dHeader.len, tstrerror(code));
      TAOS_CHECK_GOTO(code, &lino, _exit);
    }

    if (!taosCheckChecksumWhole((uint8_t *)buffer, dHeader.len)) {
      dTrace("failed to read vars body since wrong checksum");
      TAOS_CHECK_GOTO(TSDB_CODE_FILE_CORRUPTED, &lino, _exit);
    }

    ptr = buffer;
    TAOS_CHECK_GOTO(dmDecodeVars(ptr, dHeader.len, pInfo), &lino, _exit);
  }

_exit:
  if (code != 0) {
    dError("failed to read vars at line %d since %s", lino, tstrerror(code));
  }
  taosMemoryFreeClear(buffer);
  (void)taosCloseFile(&pFile);
  return code;
}

static void dmGetFname(const char *fname, char *ofname) {
  if (fname) {
    snprintf(ofname, PATH_MAX, "%s%sdnode%s%s", tsDataDir, TD_DIRSEP, TD_DIRSEP, fname);
  } else {
    snprintf(ofname, PATH_MAX, "%s%sdnode", tsDataDir, TD_DIRSEP);
  }
}

int32_t dmInitDndInfo(SDnodeData *pData) {
#ifndef _TD_DM_SKIP_CHECK
  int32_t code = 0;
  char    cfname[PATH_MAX] = "\0";

  dmGetFname(DNODE_CFG_FILE, cfname);
  bool fileExist = !(taosStatFile(cfname, NULL, NULL, NULL) < 0);
  if (fileExist) {  // dnode.info must be created before dnode.json
    return code;
  }
  dmGetFname(DM_ENGINE_FILE, cfname);
  fileExist = !(taosStatFile(cfname, NULL, NULL, NULL) < 0);
  if (fileExist) {
    return code;
  }

  int8_t      eType = 0;
  SEngineInfo eInfo = {0};
  dmFetchEType(&eType);
  eInfo.type = eType;
  eInfo.dnodeId = pData->dnodeId;
  eInfo.engineVer = tsVersion;
  eInfo.clusterId = pData->clusterId;
  eInfo.createMs = taosGetTimestampMs();
  eInfo.updateMs = eInfo.createMs;

  if ((code = dmWriteVars(&eInfo)) != 0) goto _exit;

_exit:
  TAOS_RETURN(code);
#else
  TAOS_RETURN(0);
#endif
}

static int32_t dmWriteVars(SEngineInfo *pInfo) {
  SDFHeader fHeader;
  void     *pBuf = NULL;
  void     *ptr;
  char      hbuf[DM_FILE_HEAD_SIZE] = "\0";
  char      tfname[PATH_MAX] = "\0";
  char      cfname[PATH_MAX] = "\0";
  int32_t   code = 0;
  int32_t   lino = 0;

  dmGetFname(DM_ENGINE_FILE_T, tfname);
  dmGetFname(DM_ENGINE_FILE, cfname);

  errno = 0;  // clear errno

  TdFilePtr tFile = taosOpenFile(tfname, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC);
  if (!tFile) {
    code = TAOS_SYSTEM_ERROR(errno);
    TAOS_CHECK_GOTO(code, &lino, _exit);
  }

  fHeader.version = DM_ENG_FVER_MAX;
  fHeader.len = dmEncodeVars(NULL, 0, pInfo) + sizeof(TSCKSUM);
  if (fHeader.len < 0) {
    TAOS_CHECK_GOTO(fHeader.len, &lino, _exit);
  }

  ptr = hbuf;
  TAOS_UNUSED(dmEncodeDFHeader(&ptr, &fHeader));
  TAOS_CHECK_EXIT(taosCalcChecksumAppend(0, (uint8_t *)hbuf, DM_FILE_HEAD_SIZE));

  if (taosWriteFile(tFile, hbuf, DM_FILE_HEAD_SIZE) < DM_FILE_HEAD_SIZE) {
    code = TAOS_SYSTEM_ERROR(errno);
    TAOS_CHECK_GOTO(code, &lino, _exit);
  }

  if (fHeader.len > 0) {
    if (!(pBuf = taosMemoryMalloc(fHeader.len))) {
      code = TSDB_CODE_OUT_OF_MEMORY;
      TAOS_CHECK_GOTO(code, &lino, _exit);
    }

    ptr = pBuf;
    int32_t len = dmEncodeVars(ptr, fHeader.len - sizeof(TSCKSUM), pInfo);
    if (len < 0) {
      TAOS_CHECK_GOTO(len, &lino, _exit);
    }

    TAOS_CHECK_EXIT(taosCalcChecksumAppend(0, (uint8_t *)pBuf, fHeader.len));

    if (taosWriteFile(tFile, pBuf, fHeader.len) < fHeader.len) {
      code = TAOS_SYSTEM_ERROR(errno);
      TAOS_CHECK_GOTO(code, &lino, _exit);
    }
  }

  // fsync, close and rename
  if (taosFsyncFile(tFile) < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    TAOS_CHECK_GOTO(code, &lino, _exit);
  }
  if (taosCloseFile(&tFile) < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    TAOS_CHECK_GOTO(code, &lino, _exit);
  }
  if (taosRenameFile(tfname, cfname) < 0) {
    code = TAOS_SYSTEM_ERROR(errno);
    TAOS_CHECK_GOTO(code, &lino, _exit);
  }

_exit:
  taosMemoryFreeClear(pBuf);
  if (code != 0) {
    dError("failed to write vars at line %d since %s", lino, tstrerror(code));
    TAOS_UNUSED(taosCloseFile(&tFile));
    TAOS_UNUSED(taosRemoveFile(tfname));
  }

  return code;
}

static void dmFetchEType(int8_t *type) {
  int8_t eType = DM_ETYPE_UN;
  if (STR_STR_CMP(tsVersionName, STR_STR_SIGN)) {
    if (!strncmp(tsVersionName, "t", 1)) {
      eType = DM_ETYPE_TR;
    } else {
      eType = DM_ETYPE_EN;
    }
  } else if (STR_STR_CMP(tsVersionName, STR_STR_COMM)) {
    eType = DM_ETYPE_OS;
  }
  if (type) *type = eType;
}

static int32_t dmInitVersion(SDnode *pDnode) {
#ifndef _TD_DM_SKIP_CHECK
  int32_t     code = 0;
  int32_t     lino = 0;
  int8_t      eType = 0;
  SEngineInfo eInfo = {0};

  dmFetchEType(&eType);

  char cfgFile[PATH_MAX] = "\0";
  dmGetFname(DNODE_CFG_FILE, cfgFile);

  TAOS_UNUSED(taosThreadRwlockRdlock(&pDnode->data.lock));
  // dnode.json not exist, return directly
  if (taosStatFile(cfgFile, NULL, NULL, NULL) < 0) {
    TAOS_UNUSED(taosThreadRwlockUnlock(&pDnode->data.lock));
    goto _exit;
  }
  if (((code = dmReadVars(&eInfo)) != 0) && (code != TSDB_CODE_NOT_FOUND)) {
    TAOS_UNUSED(taosThreadRwlockUnlock(&pDnode->data.lock));
    TAOS_CHECK_GOTO(code, &lino, _exit);
  }
  TAOS_UNUSED(taosThreadRwlockUnlock(&pDnode->data.lock));

  if (pDnode->data.engineVer == 0) {           // dnode.json history version
    if ((eInfo.type & 0x0F) == DM_ETYPE_UN) {  // without DM_ENGINE_FILE, create(handle update from history version)
      eInfo.type = eType;
      eInfo.dnodeId = pDnode->data.dnodeId;
      eInfo.engineVer = tsVersion;
      eInfo.clusterId = pDnode->data.clusterId;
      eInfo.createMs = taosGetTimestampMs();
      eInfo.updateMs = eInfo.createMs;
      // save
      (void)taosThreadRwlockWrlock(&pDnode->data.lock);
      if ((code = dmWriteVars(&eInfo)) != 0) {
        TAOS_UNUSED(taosThreadRwlockUnlock(&pDnode->data.lock));
        TAOS_CHECK_GOTO(code, &lino, _exit);
      }
      TAOS_UNUSED(taosThreadRwlockUnlock(&pDnode->data.lock));
      TAOS_CHECK_GOTO(dmSyncEps(&pDnode->data), &lino, _exit);
    }
  } else if ((eInfo.type & 0x0F) == DM_ETYPE_UN) {  // not history version, but without DM_ENGINE_FILE, fail
    dError("failed to init since inconsistent ver");
    TAOS_CHECK_GOTO(TSDB_CODE_VERSION_NOT_COMPATIBLE, &lino, _exit);
  } else if (pDnode->data.clusterId !=
             eInfo.clusterId) {  // not history version, DM_ENGINE_FILE exists, check clusterId
    if (eInfo.clusterId == 0) {
      eInfo.dnodeId = pDnode->data.dnodeId;
      eInfo.engineVer = tsVersion;
      eInfo.clusterId = pDnode->data.clusterId;
      eInfo.updateMs = taosGetTimestampMs();
      (void)taosThreadRwlockWrlock(&pDnode->data.lock);
      if ((code = dmWriteVars(&eInfo)) != 0) {
        TAOS_UNUSED(taosThreadRwlockUnlock(&pDnode->data.lock));
        TAOS_CHECK_GOTO(code, &lino, _exit);
      }
      TAOS_UNUSED(taosThreadRwlockUnlock(&pDnode->data.lock));
      dInfo("update clusterId from 0 to %" PRId64, pDnode->data.clusterId);
    } else {
      dError("failed to init since inconsistent cluster:%" PRIi64 ",%" PRIi64, eInfo.clusterId, pDnode->data.clusterId);
      TAOS_CHECK_GOTO(TSDB_CODE_VERSION_NOT_COMPATIBLE, &lino, _exit);
    }
  } else if (pDnode->data.engineVer != tsVersion) {  // update to latest engineVer
    TAOS_CHECK_GOTO(dmSyncEps(&pDnode->data), &lino, _exit);
  }

  if (eType == DM_ETYPE_OS) {        // oss
    if (eInfo.type > DM_ETYPE_OS) {  // enterprise to oss not allowed
      dError("failed to init since incompatible ver");
      TAOS_CHECK_GOTO(TSDB_CODE_VERSION_NOT_COMPATIBLE, &lino, _exit);
    }
  } else if (eInfo.type == DM_ETYPE_OS) {  // update oss to enterprise
    eInfo.type = eType;
    eInfo.engineVer = tsVersion;
    eInfo.updateMs = taosGetTimestampMs();
    (void)taosThreadRwlockWrlock(&pDnode->data.lock);
    if ((code = dmWriteVars(&eInfo)) != 0) {
      TAOS_UNUSED(taosThreadRwlockUnlock(&pDnode->data.lock));
      TAOS_CHECK_GOTO(code, &lino, _exit);
    }
    TAOS_UNUSED(taosThreadRwlockUnlock(&pDnode->data.lock));
  }

_exit:
  if (code != 0) {
    dError("failed to init version at line %d since %s", lino, tstrerror(code));
  }
  TAOS_RETURN(code);
#else
  TAOS_RETURN(0);
#endif
}

static int32_t dmSyncEps(SDnodeData *pData) {
  int32_t code = 0;
  char    file[PATH_MAX] = "\0";
  (void)snprintf(file, sizeof(file), "%s%sdnode%sdnode.json", tsDataDir, TD_DIRSEP, TD_DIRSEP);
  (void)taosThreadRwlockWrlock(&pData->lock);
  bool fileExist = !(taosStatFile(file, NULL, NULL, NULL) < 0);
  if (fileExist) {
    code = dmWriteEps(pData);
  }
  TAOS_UNUSED(taosThreadRwlockUnlock(&pData->lock));
  TAOS_RETURN(code);
}

// invoker
int32_t dmInitModule(SDnode *pDnode) {
  int32_t code = 0;
  int32_t lino = 0;

#ifdef _TD_DM_CHECK_OFFSET
  TAOS_CHECK_GOTO(dmCheckOffset(pDnode), &lino, _exit);
#endif

  TAOS_CHECK_GOTO(dmInitPrerequisites(), &lino, _exit);

  if (dmInitVersion(pDnode) != 0) {
    TAOS_CHECK_GOTO(TSDB_CODE_VERSION_NOT_COMPATIBLE, &lino, _exit);
  }

  TAOS_CHECK_GOTO(dmInitMsgHandle(pDnode), &lino, _exit);

  TAOS_CHECK_GOTO(dmInitServer(pDnode), &lino, _exit);

  TAOS_CHECK_GOTO(dmInitClient(pDnode), &lino, _exit);

_exit:
  if (code != 0) {
    dError("failed to init module at line %d since %s", lino, tstrerror(code));
  }

  TAOS_RETURN(code);
}

1	/*
2	* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
3	*
4	* This program is free software: you can use, redistribute, and/or modify
5	* it under the terms of the GNU Affero General Public License, version 3
6	* or later ("AGPL"), as published by the Free Software Foundation.
7	*
8	* This program is distributed in the hope that it will be useful, but WITHOUT
9	* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
10	* FITNESS FOR A PARTICULAR PURPOSE.
11	*
12	* You should have received a copy of the GNU Affero General Public License
13	* along with this program. If not, see <http://www.gnu.org/licenses/>.
14	*/
15
16	#define _DEFAULT_SOURCE
17
18	#include "dmMgmt.h"
19	#include "tchecksum.h"
20
21	#if defined(GRANTS_CFG) \|\| defined(_TD_DARWIN_64)
22	#define _TD_DM_SKIP_CHECK
23	#endif
24
25	#if !defined(ASSERT_NOT_CORE) && !defined(WINDOWS)
26	#define _TD_DM_CHECK_OFFSET
27	#define DM_CHECK_OFFSET(p1, p2, offset, flag) \
28	do { \
29	int32_t off = POINTER_DISTANCE((p1), (p2)); \
30	if ((offset) != abs(off)) { \
31	dError("%s offset: %d!=%d", (flag), off, (offset)); \
32	return TSDB_CODE_INTERNAL_ERROR; \
33	} \
34	} while (0)
35	#endif
36
37	extern char tsVersionName[];
38
39	typedef enum {
40	DM_ENG_FVER_1 = 1,
41	} DM_ENG_FVER;
42
43	typedef enum {
44	DM_ETYPE_UN = 0,
45	DM_ETYPE_OS = 1,
46	DM_ETYPE_TR = 2,
47	DM_ETYPE_EN = 3,
48	} DM_ENG_TYPE;
49
50	#define DM_ENG_FVER_MAX DM_ENG_FVER_1
51	#define DM_OS_ST_NAME_LEN 64
52
53	#define DM_ENGINE_FILE "dnode.info"
54	#define DM_ENGINE_FILE_T "dnode.info.t"
55	#define DNODE_CFG_FILE "dnode.json"
56
57	typedef struct {
58	int8_t type; // 0 unknown 1 community 2 trial 3 official
59	int32_t dnodeId;
60	int32_t engineVer; // tsVersion
61	int64_t clusterId;
62	int64_t createMs;
63	int64_t updateMs;
64	} SEngineInfo;
65
66	// file operation (TODO: refactor to dmUtil.c)
67	#define DM_FILE_HEAD_SIZE 512
68	typedef struct {
69	uint32_t version;
70	uint32_t len; // Encoded content len(checksum included)
71	} SDFHeader;
72
73	#define STR_CASE_CMP(s, d) (0 == strcasecmp((s), (d)))
74	#define STR_STR_CMP(s, d) (strstr((s), (d)))
75	#define STR_INT_CMP(s, d, c) (taosStr2Int32(s, 0, 10) c(d))
76	#define STR_STR_SIGN ("ia")
77	#define STR_STR_COMM ("unit")
78
79	#define STR_CASE_STR_CHECK(s, d) \
80	do { \
81	TAOS_UNUSED(strtolower(s, s)); \
82	TAOS_UNUSED(strtolower(d, d)); \
83	if (STR_STR_CMP(s, d)) { \
84	DM_ERR_RTN(0); \
85	} \
86	} while (0)
87
88	#define DM_ERR_RTN(c) \
89	do { \
90	code = (c); \
91	goto _exit; \
92	} while (0)
93
94	static const char *dmOS[] = {"Ubuntu", "CentOS Linux", "Red Hat", "Debian GNU/Linux", "CoreOS", "FreeBSD", "openSUSE",
95	"SLES", "Fedora", "macOS", "CentOS Stream"};
96
97	// declarations
98	static void dmFetchEType(int8_t *type);
99	static void dmGetFname(const char fname, char ofname);
100	static int32_t dmSyncEps(SDnodeData *pData);
101	static int32_t dmEncodeVars(void buf, int32_t bufLen, SEngineInfo pInfo);
102	static int32_t dmEncodeVars(void buf, int32_t bufLen, SEngineInfo pInfo);
103	static int32_t dmReadVars(SEngineInfo *pInfo);
104	static int32_t dmWriteVars(SEngineInfo *pInfo);
105
106	// implementations
107
108	#ifdef _TD_DM_CHECK_OFFSET
109	static int32_t dmCheckOffset(SDnode *pDnode) {	135✔
110	SDnodeData *pData = &pDnode->data;	135✔
111	TdThreadRwlock *pLock = &pData->lock;	135✔
112	int32_t *pDnodeId = &pData->dnodeId;	135✔
113	int32_t *pEngineVer = &pData->engineVer;	135✔
114	int64_t *pClusterId = &pData->clusterId;	135✔
115
116	DM_CHECK_OFFSET(pData, pDnode, 168, "dnode data");	135!
117	DM_CHECK_OFFSET(pLock, pData, 720, "data lock");	135!
118	DM_CHECK_OFFSET(pDnodeId, pData, 0, "data dnodeId");	135!
119	DM_CHECK_OFFSET(pEngineVer, pData, 4, "data engineVer");	135!
120	DM_CHECK_OFFSET(pClusterId, pData, 8, "data clusterId");	135!
121
122	return TSDB_CODE_SUCCESS;	135✔
123	}
124	#endif
125
126	static int32_t dmInitPrerequisites() {	135✔
127	#ifndef _TD_DM_SKIP_CHECK
128	int32_t code = 0;	135✔
129
130	char reName[64] = {0};	135✔
131	char stName[64] = {0};	135✔
132	char ver[64] = {0};	135✔
133
134	code = (int32_t)(2147483648 \| 298);	135✔
135	tstrncpy(stName, tsVersionName, 16);	135✔
136
137	if (STR_STR_CMP(stName, STR_STR_SIGN)) {	135!
138	DM_ERR_RTN(0);	135✔
139	}
140	if (taosGetOsReleaseName(reName, stName, ver, DM_OS_ST_NAME_LEN) != 0) {	×
141	int32_t errCode = TAOS_SYSTEM_ERROR(errno);	×
142	if (errCode != 0) code = errCode;	×
143	TAOS_CHECK_GOTO(code, NULL, _exit);	×
144	}
145	if (STR_CASE_CMP(stName, dmOS[0])) {	×
146	if (STR_INT_CMP(ver, 17, >)) {	×
147	DM_ERR_RTN(0);	×
148	}
149	} else if (STR_CASE_CMP(stName, dmOS[1])) {	×
150	if (STR_INT_CMP(ver, 6, >)) {	×
151	DM_ERR_RTN(0);	×
152	}
153	} else {
154	int32_t size = sizeof(dmOS) / sizeof(dmOS[0]);	×
155	char os[DM_OS_ST_NAME_LEN] = {0};	×
156	for (int32_t i = 2; i < size; ++i) {	×
157	tstrncpy(os, dmOS[i], DM_OS_ST_NAME_LEN);	×
158	STR_CASE_STR_CHECK(stName, os);	×
159	}
160	}
161
162	_exit:	135✔
163	TAOS_RETURN(code);	135✔
164	#else
165	TAOS_RETURN(0);
166	#endif
167	}
168
169	static int dmEncodeDFHeader(void *buf, SDFHeader pHeader) {	60✔
170	int tlen = 0;	60✔
171
172	tlen += taosEncodeFixedU32(buf, pHeader->version);	60!
173	tlen += taosEncodeFixedU32(buf, pHeader->len);	60!
174
175	return tlen;	60✔
176	}
177
178	static void dmDecodeDFHeader(void buf, SDFHeader *pHeader) {	23✔
179	buf = taosDecodeFixedU32(buf, &(pHeader->version));	23!
180	buf = taosDecodeFixedU32(buf, &(pHeader->len));	23!
181
182	return buf;	23✔
183	}
184
185	static int32_t dmEncodeVars(void buf, int32_t bufLen, SEngineInfo pInfo) {	120✔
186	SEncoder encoder = {0};	120✔
187	tEncoderInit(&encoder, buf, bufLen);	120✔
188	int32_t code = 0;	120✔
189
190	TAOS_CHECK_GOTO(tStartEncode(&encoder), NULL, _exit);	120!
191	TAOS_CHECK_GOTO(tEncodeI8(&encoder, pInfo->type), NULL, _exit);	240!
192	TAOS_CHECK_GOTO(tEncodeI32v(&encoder, pInfo->dnodeId), NULL, _exit);	240!
193	TAOS_CHECK_GOTO(tEncodeI32v(&encoder, pInfo->engineVer), NULL, _exit);	240!
194	TAOS_CHECK_GOTO(tEncodeI64v(&encoder, pInfo->clusterId), NULL, _exit);	240!
195	TAOS_CHECK_GOTO(tEncodeI64v(&encoder, pInfo->createMs), NULL, _exit);	240!
196	TAOS_CHECK_GOTO(tEncodeI64v(&encoder, pInfo->updateMs), NULL, _exit);	240!
197
198	tEndEncode(&encoder);	120✔
199	code = encoder.pos;	120✔
200	_exit:	120✔
201	tEncoderClear(&encoder);	120✔
202	return code;	120✔
203	}
204
205	static int32_t dmDecodeVars(void buf, int32_t bufLen, SEngineInfo pInfo) {	23✔
206	SDecoder decoder = {0};	23✔
207	tDecoderInit(&decoder, buf, bufLen);	23✔
208	int32_t code = 0;	23✔
209
210	TAOS_CHECK_GOTO(tStartDecode(&decoder), NULL, _exit);	23!
211
212	TAOS_CHECK_GOTO(tDecodeI8(&decoder, &pInfo->type), NULL, _exit);	46!
213	TAOS_CHECK_GOTO(tDecodeI32v(&decoder, &pInfo->dnodeId), NULL, _exit);	46!
214	TAOS_CHECK_GOTO(tDecodeI32v(&decoder, &pInfo->engineVer), NULL, _exit);	46!
215	TAOS_CHECK_GOTO(tDecodeI64v(&decoder, &pInfo->clusterId), NULL, _exit);	46!
216	TAOS_CHECK_GOTO(tDecodeI64v(&decoder, &pInfo->createMs), NULL, _exit);	46!
217	TAOS_CHECK_GOTO(tDecodeI64v(&decoder, &pInfo->updateMs), NULL, _exit);	46!
218
219	tEndDecode(&decoder);	23✔
220
221	_exit:	23✔
222	tDecoderClear(&decoder);	23✔
223	TAOS_RETURN(code);	23✔
224	}
225
226	static int32_t dmReadVars(SEngineInfo *pInfo) {	23✔
227	int32_t code = 0;	23✔
228	int32_t lino = 0;	23✔
229	TdFilePtr pFile = NULL;	23✔
230	void *buffer = NULL;	23✔
231	void *ptr;
232	SDFHeader dHeader;
233	char fname[FILENAME_MAX] = "\0";	23✔
234
235	dmGetFname(DM_ENGINE_FILE, fname);	23✔
236
237	errno = 0; // clear errno	23✔
238
239	if (!taosCheckExistFile(fname)) {	23!
240	TAOS_CHECK_GOTO(TSDB_CODE_NOT_FOUND, &lino, _exit);	×
241	}
242
243	pFile = taosOpenFile(fname, TD_FILE_READ);	23✔
244	if (!pFile) {	23!
245	if (errno == ENOENT) {	×
246	TAOS_CHECK_GOTO(TSDB_CODE_NOT_FOUND, &lino, _exit);	×
247	} else {
248	TAOS_CHECK_GOTO(TAOS_SYSTEM_ERROR(errno), &lino, _exit);	×
249	}
250	}
251
252	if (!(buffer = taosMemoryMalloc(DM_FILE_HEAD_SIZE))) {	23!
253	TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, &lino, _exit);	×
254	}
255
256	int64_t nRead = taosReadFile(pFile, buffer, DM_FILE_HEAD_SIZE);	23✔
257	if (nRead < 0) {	23!
258	TAOS_CHECK_GOTO(TAOS_SYSTEM_ERROR(errno), &lino, _exit);	×
259	}
260
261	if (nRead != DM_FILE_HEAD_SIZE) {	23!
262	code = TSDB_CODE_FILE_CORRUPTED;	×
263	dTrace("failed to read %d bytes from vars head since %s", DM_FILE_HEAD_SIZE, tstrerror(code));	×
264	TAOS_CHECK_GOTO(code, &lino, _exit);	×
265	}
266
267	if (!taosCheckChecksumWhole((uint8_t *)buffer, DM_FILE_HEAD_SIZE)) {	23!
268	dTrace("failed to read vars head since wrong checksum");	×
269	TAOS_CHECK_GOTO(TSDB_CODE_CHECKSUM_ERROR, &lino, _exit);	×
270	}
271
272	ptr = buffer;	23✔
273	ptr = dmDecodeDFHeader(ptr, &dHeader);	23✔
274
275	if (dHeader.version != DM_ENG_FVER_1) {	23✔
276	// TODO
277	}
278
279	if (dHeader.len > 0) {	23!
280	if (dHeader.len > DM_FILE_HEAD_SIZE) {	23!
281	void *tmpBuf = NULL;	×
282	if (!(tmpBuf = taosMemoryRealloc(buffer, dHeader.len))) {	×
283	TAOS_CHECK_GOTO(TSDB_CODE_OUT_OF_MEMORY, &lino, _exit);	×
284	}
285	buffer = tmpBuf;	×
286	}
287
288	nRead = (int)taosReadFile(pFile, buffer, dHeader.len);	23✔
289	if (nRead < 0) {	23!
290	code = TAOS_SYSTEM_ERROR(errno);	×
291	TAOS_CHECK_GOTO(code, &lino, _exit);	×
292	}
293
294	if (nRead != dHeader.len) {	23!
295	code = TSDB_CODE_FILE_CORRUPTED;	×
296	dTrace("failed to read %d bytes from vars body since %s", dHeader.len, tstrerror(code));	×
297	TAOS_CHECK_GOTO(code, &lino, _exit);	×
298	}
299
300	if (!taosCheckChecksumWhole((uint8_t *)buffer, dHeader.len)) {	46!
301	dTrace("failed to read vars body since wrong checksum");	×
302	TAOS_CHECK_GOTO(TSDB_CODE_FILE_CORRUPTED, &lino, _exit);	×
303	}
304
305	ptr = buffer;	23✔
306	TAOS_CHECK_GOTO(dmDecodeVars(ptr, dHeader.len, pInfo), &lino, _exit);	23!
307	}
308
309	_exit:	23✔
310	if (code != 0) {	23!
311	dError("failed to read vars at line %d since %s", lino, tstrerror(code));	×
312	}
313	taosMemoryFreeClear(buffer);	23!
314	(void)taosCloseFile(&pFile);	23✔
315	return code;	23✔
316	}
317
318	static void dmGetFname(const char fname, char ofname) {	587✔
319	if (fname) {	587!
320	snprintf(ofname, PATH_MAX, "%s%sdnode%s%s", tsDataDir, TD_DIRSEP, TD_DIRSEP, fname);	587✔
321	} else {
322	snprintf(ofname, PATH_MAX, "%s%sdnode", tsDataDir, TD_DIRSEP);	×
323	}
324	}	587✔
325
326	int32_t dmInitDndInfo(SDnodeData *pData) {	250✔
327	#ifndef _TD_DM_SKIP_CHECK
328	int32_t code = 0;	250✔
329	char cfname[PATH_MAX] = "\0";	250✔
330
331	dmGetFname(DNODE_CFG_FILE, cfname);	250✔
332	bool fileExist = !(taosStatFile(cfname, NULL, NULL, NULL) < 0);	250✔
333	if (fileExist) { // dnode.info must be created before dnode.json	250✔
334	return code;	191✔
335	}
336	dmGetFname(DM_ENGINE_FILE, cfname);	59✔
337	fileExist = !(taosStatFile(cfname, NULL, NULL, NULL) < 0);	59✔
338	if (fileExist) {	59!
339	return code;	×
340	}
341
342	int8_t eType = 0;	59✔
343	SEngineInfo eInfo = {0};	59✔
344	dmFetchEType(&eType);	59✔
345	eInfo.type = eType;	59✔
346	eInfo.dnodeId = pData->dnodeId;	59✔
347	eInfo.engineVer = tsVersion;	59✔
348	eInfo.clusterId = pData->clusterId;	59✔
349	eInfo.createMs = taosGetTimestampMs();	59✔
350	eInfo.updateMs = eInfo.createMs;	59✔
351
352	if ((code = dmWriteVars(&eInfo)) != 0) goto _exit;	59!
353
354	_exit:	59✔
355	TAOS_RETURN(code);	59✔
356	#else
357	TAOS_RETURN(0);
358	#endif
359	}
360
361	static int32_t dmWriteVars(SEngineInfo *pInfo) {	60✔
362	SDFHeader fHeader;
363	void *pBuf = NULL;	60✔
364	void *ptr;
365	char hbuf[DM_FILE_HEAD_SIZE] = "\0";	60✔
366	char tfname[PATH_MAX] = "\0";	60✔
367	char cfname[PATH_MAX] = "\0";	60✔
368	int32_t code = 0;	60✔
369	int32_t lino = 0;	60✔
370
371	dmGetFname(DM_ENGINE_FILE_T, tfname);	60✔
372	dmGetFname(DM_ENGINE_FILE, cfname);	60✔
373
374	errno = 0; // clear errno	60✔
375
376	TdFilePtr tFile = taosOpenFile(tfname, TD_FILE_CREATE \| TD_FILE_WRITE \| TD_FILE_TRUNC);	60✔
377	if (!tFile) {	60!
378	code = TAOS_SYSTEM_ERROR(errno);	×
379	TAOS_CHECK_GOTO(code, &lino, _exit);	×
380	}
381
382	fHeader.version = DM_ENG_FVER_MAX;	60✔
383	fHeader.len = dmEncodeVars(NULL, 0, pInfo) + sizeof(TSCKSUM);	60✔
384	if (fHeader.len < 0) {
385	TAOS_CHECK_GOTO(fHeader.len, &lino, _exit);
386	}
387
388	ptr = hbuf;	60✔
389	TAOS_UNUSED(dmEncodeDFHeader(&ptr, &fHeader));	60✔
390	TAOS_CHECK_EXIT(taosCalcChecksumAppend(0, (uint8_t *)hbuf, DM_FILE_HEAD_SIZE));	60!
391
392	if (taosWriteFile(tFile, hbuf, DM_FILE_HEAD_SIZE) < DM_FILE_HEAD_SIZE) {	60!
393	code = TAOS_SYSTEM_ERROR(errno);	×
394	TAOS_CHECK_GOTO(code, &lino, _exit);	×
395	}
396
397	if (fHeader.len > 0) {	60!
398	if (!(pBuf = taosMemoryMalloc(fHeader.len))) {	60!
399	code = TSDB_CODE_OUT_OF_MEMORY;	×
400	TAOS_CHECK_GOTO(code, &lino, _exit);	×
401	}
402
403	ptr = pBuf;	60✔
404	int32_t len = dmEncodeVars(ptr, fHeader.len - sizeof(TSCKSUM), pInfo);	60✔
405	if (len < 0) {	60!
406	TAOS_CHECK_GOTO(len, &lino, _exit);	×
407	}
408
409	TAOS_CHECK_EXIT(taosCalcChecksumAppend(0, (uint8_t *)pBuf, fHeader.len));	120!
410
411	if (taosWriteFile(tFile, pBuf, fHeader.len) < fHeader.len) {	60!
412	code = TAOS_SYSTEM_ERROR(errno);	×
413	TAOS_CHECK_GOTO(code, &lino, _exit);	×
414	}
415	}
416
417	// fsync, close and rename
418	if (taosFsyncFile(tFile) < 0) {	60!
419	code = TAOS_SYSTEM_ERROR(errno);	×
420	TAOS_CHECK_GOTO(code, &lino, _exit);	×
421	}
422	if (taosCloseFile(&tFile) < 0) {	60!
423	code = TAOS_SYSTEM_ERROR(errno);	×
424	TAOS_CHECK_GOTO(code, &lino, _exit);	×
425	}
426	if (taosRenameFile(tfname, cfname) < 0) {	60!
427	code = TAOS_SYSTEM_ERROR(errno);	×
428	TAOS_CHECK_GOTO(code, &lino, _exit);	×
429	}
430
431	_exit:	60✔
432	taosMemoryFreeClear(pBuf);	60!
433	if (code != 0) {	60!
434	dError("failed to write vars at line %d since %s", lino, tstrerror(code));	×
435	TAOS_UNUSED(taosCloseFile(&tFile));	×
436	TAOS_UNUSED(taosRemoveFile(tfname));	×
437	}
438
439	return code;	60✔
440	}
441
442	static void dmFetchEType(int8_t *type) {	194✔
443	int8_t eType = DM_ETYPE_UN;	194✔
444	if (STR_STR_CMP(tsVersionName, STR_STR_SIGN)) {	194!
445	if (!strncmp(tsVersionName, "t", 1)) {	194!
446	eType = DM_ETYPE_TR;	194✔
447	} else {
448	eType = DM_ETYPE_EN;	×
449	}
450	} else if (STR_STR_CMP(tsVersionName, STR_STR_COMM)) {	×
451	eType = DM_ETYPE_OS;	×
452	}
453	if (type) *type = eType;	194!
454	}	194✔
455
456	static int32_t dmInitVersion(SDnode *pDnode) {	135✔
457	#ifndef _TD_DM_SKIP_CHECK
458	int32_t code = 0;	135✔
459	int32_t lino = 0;	135✔
460	int8_t eType = 0;	135✔
461	SEngineInfo eInfo = {0};	135✔
462
463	dmFetchEType(&eType);	135✔
464
465	char cfgFile[PATH_MAX] = "\0";	135✔
466	dmGetFname(DNODE_CFG_FILE, cfgFile);	135✔
467
468	TAOS_UNUSED(taosThreadRwlockRdlock(&pDnode->data.lock));	135✔
469	// dnode.json not exist, return directly
470	if (taosStatFile(cfgFile, NULL, NULL, NULL) < 0) {	135✔
471	TAOS_UNUSED(taosThreadRwlockUnlock(&pDnode->data.lock));	112✔
472	goto _exit;	112✔
473	}
474	if (((code = dmReadVars(&eInfo)) != 0) && (code != TSDB_CODE_NOT_FOUND)) {	23!
475	TAOS_UNUSED(taosThreadRwlockUnlock(&pDnode->data.lock));	×
476	TAOS_CHECK_GOTO(code, &lino, _exit);	×
477	}
478	TAOS_UNUSED(taosThreadRwlockUnlock(&pDnode->data.lock));	23✔
479
480	if (pDnode->data.engineVer == 0) { // dnode.json history version	23!
481	if ((eInfo.type & 0x0F) == DM_ETYPE_UN) { // without DM_ENGINE_FILE, create(handle update from history version)	×
482	eInfo.type = eType;	×
483	eInfo.dnodeId = pDnode->data.dnodeId;	×
484	eInfo.engineVer = tsVersion;	×
485	eInfo.clusterId = pDnode->data.clusterId;	×
486	eInfo.createMs = taosGetTimestampMs();	×
487	eInfo.updateMs = eInfo.createMs;	×
488	// save
489	(void)taosThreadRwlockWrlock(&pDnode->data.lock);	×
490	if ((code = dmWriteVars(&eInfo)) != 0) {	×
491	TAOS_UNUSED(taosThreadRwlockUnlock(&pDnode->data.lock));	×
492	TAOS_CHECK_GOTO(code, &lino, _exit);	×
493	}
494	TAOS_UNUSED(taosThreadRwlockUnlock(&pDnode->data.lock));	×
495	TAOS_CHECK_GOTO(dmSyncEps(&pDnode->data), &lino, _exit);	×
496	}
497	} else if ((eInfo.type & 0x0F) == DM_ETYPE_UN) { // not history version, but without DM_ENGINE_FILE, fail	23!
498	dError("failed to init since inconsistent ver");	×
499	TAOS_CHECK_GOTO(TSDB_CODE_VERSION_NOT_COMPATIBLE, &lino, _exit);	×
500	} else if (pDnode->data.clusterId !=	23✔
501	eInfo.clusterId) { // not history version, DM_ENGINE_FILE exists, check clusterId	23!
502	if (eInfo.clusterId == 0) {	×
503	eInfo.dnodeId = pDnode->data.dnodeId;	×
504	eInfo.engineVer = tsVersion;	×
505	eInfo.clusterId = pDnode->data.clusterId;	×
506	eInfo.updateMs = taosGetTimestampMs();	×
507	(void)taosThreadRwlockWrlock(&pDnode->data.lock);	×
508	if ((code = dmWriteVars(&eInfo)) != 0) {	×
509	TAOS_UNUSED(taosThreadRwlockUnlock(&pDnode->data.lock));	×
510	TAOS_CHECK_GOTO(code, &lino, _exit);	×
511	}
512	TAOS_UNUSED(taosThreadRwlockUnlock(&pDnode->data.lock));	×
513	dInfo("update clusterId from 0 to %" PRId64, pDnode->data.clusterId);	×
514	} else {
515	dError("failed to init since inconsistent cluster:%" PRIi64 ",%" PRIi64, eInfo.clusterId, pDnode->data.clusterId);	×
516	TAOS_CHECK_GOTO(TSDB_CODE_VERSION_NOT_COMPATIBLE, &lino, _exit);	×
517	}
518	} else if (pDnode->data.engineVer != tsVersion) { // update to latest engineVer	23!
519	TAOS_CHECK_GOTO(dmSyncEps(&pDnode->data), &lino, _exit);	×
520	}
521
522	if (eType == DM_ETYPE_OS) { // oss	23!
523	if (eInfo.type > DM_ETYPE_OS) { // enterprise to oss not allowed	×
524	dError("failed to init since incompatible ver");	×
525	TAOS_CHECK_GOTO(TSDB_CODE_VERSION_NOT_COMPATIBLE, &lino, _exit);	×
526	}
527	} else if (eInfo.type == DM_ETYPE_OS) { // update oss to enterprise	23✔
528	eInfo.type = eType;	1✔
529	eInfo.engineVer = tsVersion;	1✔
530	eInfo.updateMs = taosGetTimestampMs();	1✔
531	(void)taosThreadRwlockWrlock(&pDnode->data.lock);	1✔
532	if ((code = dmWriteVars(&eInfo)) != 0) {	1!
533	TAOS_UNUSED(taosThreadRwlockUnlock(&pDnode->data.lock));	×
534	TAOS_CHECK_GOTO(code, &lino, _exit);	×
535	}
536	TAOS_UNUSED(taosThreadRwlockUnlock(&pDnode->data.lock));	1✔
537	}
538
539	_exit:	22✔
540	if (code != 0) {	135!
541	dError("failed to init version at line %d since %s", lino, tstrerror(code));	×
542	}
543	TAOS_RETURN(code);	135✔
544	#else
545	TAOS_RETURN(0);
546	#endif
547	}
548
549	static int32_t dmSyncEps(SDnodeData *pData) {	×
550	int32_t code = 0;	×
551	char file[PATH_MAX] = "\0";	×
552	(void)snprintf(file, sizeof(file), "%s%sdnode%sdnode.json", tsDataDir, TD_DIRSEP, TD_DIRSEP);	×
553	(void)taosThreadRwlockWrlock(&pData->lock);	×
554	bool fileExist = !(taosStatFile(file, NULL, NULL, NULL) < 0);	×
555	if (fileExist) {	×
556	code = dmWriteEps(pData);	×
557	}
558	TAOS_UNUSED(taosThreadRwlockUnlock(&pData->lock));	×
559	TAOS_RETURN(code);	×
560	}
561
562	// invoker
563	int32_t dmInitModule(SDnode *pDnode) {	135✔
564	int32_t code = 0;	135✔
565	int32_t lino = 0;	135✔
566
567	#ifdef _TD_DM_CHECK_OFFSET
568	TAOS_CHECK_GOTO(dmCheckOffset(pDnode), &lino, _exit);	135!
569	#endif
570
571	TAOS_CHECK_GOTO(dmInitPrerequisites(), &lino, _exit);	135!
572
573	if (dmInitVersion(pDnode) != 0) {	135!
574	TAOS_CHECK_GOTO(TSDB_CODE_VERSION_NOT_COMPATIBLE, &lino, _exit);	×
575	}
576
577	TAOS_CHECK_GOTO(dmInitMsgHandle(pDnode), &lino, _exit);	135!
578
579	TAOS_CHECK_GOTO(dmInitServer(pDnode), &lino, _exit);	135✔
580
581	TAOS_CHECK_GOTO(dmInitClient(pDnode), &lino, _exit);	133!
582
583	_exit:	133✔
584	if (code != 0) {	135✔
585	dError("failed to init module at line %d since %s", lino, tstrerror(code));	2!
586	}
587
588	TAOS_RETURN(code);	135✔
589	}

taosdata / TDengine / #3834

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous