#4361

Committed 25 Jun 2025 09:25AM UTC coverage: 62.074% (-0.4%) from 62.521%

Build # #4361

Build Type

push

travis-ci

Committed by GitHub

Commit Message

docs: add OpenMetrics support and configuration details to taosAdapter documentation (#31427)

Run Details

155314 of 319947 branches covered (48.54%)

Branch coverage included in aggregate %.

241251 of 318911 relevant lines covered (75.65%)

6190992.11 hits per line

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

67.93

/source/dnode/mgmt/mgmt_vnode/src/vmFile.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 "tjson.h"
#include "vmInt.h"

#define MAX_CONTENT_LEN 2 * 1024 * 1024

int32_t vmGetAllVnodeListFromHash(SVnodeMgmt *pMgmt, int32_t *numOfVnodes, SVnodeObj ***ppVnodes) {
  (void)taosThreadRwlockRdlock(&pMgmt->hashLock);

  int32_t num = 0;
  int32_t size = taosHashGetSize(pMgmt->runngingHash);
  int32_t closedSize = taosHashGetSize(pMgmt->closedHash);
  size += closedSize;
  SVnodeObj **pVnodes = taosMemoryCalloc(size, sizeof(SVnodeObj *));
  if (pVnodes == NULL) {
    (void)taosThreadRwlockUnlock(&pMgmt->hashLock);
    return terrno;
  }

  void *pIter = taosHashIterate(pMgmt->runngingHash, NULL);
  while (pIter) {
    SVnodeObj **ppVnode = pIter;
    SVnodeObj  *pVnode = *ppVnode;
    if (pVnode && num < size) {
      int32_t refCount = atomic_add_fetch_32(&pVnode->refCount, 1);
      dTrace("vgId:%d,acquire vnode, vnode:%p, ref:%d", pVnode->vgId, pVnode, refCount);
      pVnodes[num++] = (*ppVnode);
      pIter = taosHashIterate(pMgmt->runngingHash, pIter);
    } else {
      taosHashCancelIterate(pMgmt->runngingHash, pIter);
    }
  }

  pIter = taosHashIterate(pMgmt->closedHash, NULL);
  while (pIter) {
    SVnodeObj **ppVnode = pIter;
    SVnodeObj  *pVnode = *ppVnode;
    if (pVnode && num < size) {
      int32_t refCount = atomic_add_fetch_32(&pVnode->refCount, 1);
      dTrace("vgId:%d, acquire vnode, vnode:%p, ref:%d", pVnode->vgId, pVnode, refCount);
      pVnodes[num++] = (*ppVnode);
      pIter = taosHashIterate(pMgmt->closedHash, pIter);
    } else {
      taosHashCancelIterate(pMgmt->closedHash, pIter);
    }
  }

  (void)taosThreadRwlockUnlock(&pMgmt->hashLock);
  *numOfVnodes = num;
  *ppVnodes = pVnodes;

  return 0;
}

int32_t vmGetAllVnodeListFromHashWithCreating(SVnodeMgmt *pMgmt, int32_t *numOfVnodes, SVnodeObj ***ppVnodes) {
  (void)taosThreadRwlockRdlock(&pMgmt->hashLock);

  int32_t num = 0;
  int32_t size = taosHashGetSize(pMgmt->runngingHash);
  int32_t creatingSize = taosHashGetSize(pMgmt->creatingHash);
  size += creatingSize;
  SVnodeObj **pVnodes = taosMemoryCalloc(size, sizeof(SVnodeObj *));
  if (pVnodes == NULL) {
    (void)taosThreadRwlockUnlock(&pMgmt->hashLock);
    return terrno;
  }

  void *pIter = taosHashIterate(pMgmt->runngingHash, NULL);
  while (pIter) {
    SVnodeObj **ppVnode = pIter;
    SVnodeObj  *pVnode = *ppVnode;
    if (pVnode && num < size) {
      int32_t refCount = atomic_add_fetch_32(&pVnode->refCount, 1);
      dTrace("vgId:%d,acquire vnode, vnode:%p, ref:%d", pVnode->vgId, pVnode, refCount);
      pVnodes[num++] = (*ppVnode);
      pIter = taosHashIterate(pMgmt->runngingHash, pIter);
    } else {
      taosHashCancelIterate(pMgmt->runngingHash, pIter);
    }
  }

  pIter = taosHashIterate(pMgmt->creatingHash, NULL);
  while (pIter) {
    SVnodeObj **ppVnode = pIter;
    SVnodeObj  *pVnode = *ppVnode;
    if (pVnode && num < size) {
      int32_t refCount = atomic_add_fetch_32(&pVnode->refCount, 1);
      dTrace("vgId:%d, acquire vnode, vnode:%p, ref:%d", pVnode->vgId, pVnode, refCount);
      pVnodes[num++] = (*ppVnode);
      pIter = taosHashIterate(pMgmt->creatingHash, pIter);
    } else {
      taosHashCancelIterate(pMgmt->creatingHash, pIter);
    }
  }
  (void)taosThreadRwlockUnlock(&pMgmt->hashLock);

  *numOfVnodes = num;
  *ppVnodes = pVnodes;

  return 0;
}

int32_t vmGetVnodeListFromHash(SVnodeMgmt *pMgmt, int32_t *numOfVnodes, SVnodeObj ***ppVnodes) {
  (void)taosThreadRwlockRdlock(&pMgmt->hashLock);

  int32_t     num = 0;
  int32_t     size = taosHashGetSize(pMgmt->runngingHash);
  SVnodeObj **pVnodes = taosMemoryCalloc(size, sizeof(SVnodeObj *));
  if (pVnodes == NULL) {
    (void)taosThreadRwlockUnlock(&pMgmt->hashLock);
    return terrno;
  }

  void *pIter = taosHashIterate(pMgmt->runngingHash, NULL);
  while (pIter) {
    SVnodeObj **ppVnode = pIter;
    SVnodeObj  *pVnode = *ppVnode;
    if (pVnode && num < size) {
      int32_t refCount = atomic_add_fetch_32(&pVnode->refCount, 1);
      dTrace("vgId:%d, acquire vnode, vnode:%p, ref:%d", pVnode->vgId, pVnode, refCount);
      pVnodes[num++] = (*ppVnode);
      pIter = taosHashIterate(pMgmt->runngingHash, pIter);
    } else {
      taosHashCancelIterate(pMgmt->runngingHash, pIter);
    }
  }

  (void)taosThreadRwlockUnlock(&pMgmt->hashLock);
  *numOfVnodes = num;
  *ppVnodes = pVnodes;

  return 0;
}

static int32_t vmDecodeVnodeList(SJson *pJson, SVnodeMgmt *pMgmt, SWrapperCfg **ppCfgs, int32_t *numOfVnodes) {
  int32_t      code = -1;
  SWrapperCfg *pCfgs = NULL;
  *ppCfgs = NULL;

  SJson *vnodes = tjsonGetObjectItem(pJson, "vnodes");
  if (vnodes == NULL) return TSDB_CODE_INVALID_JSON_FORMAT;

  int32_t vnodesNum = cJSON_GetArraySize(vnodes);
  if (vnodesNum > 0) {
    pCfgs = taosMemoryCalloc(vnodesNum, sizeof(SWrapperCfg));
    if (pCfgs == NULL) return terrno;
  }

  for (int32_t i = 0; i < vnodesNum; ++i) {
    SJson *vnode = tjsonGetArrayItem(vnodes, i);
    if (vnode == NULL) {
      code = TSDB_CODE_INVALID_JSON_FORMAT;
      goto _OVER;
    }

    SWrapperCfg *pCfg = &pCfgs[i];
    tjsonGetInt32ValueFromDouble(vnode, "vgId", pCfg->vgId, code);
    if (code != 0) goto _OVER;
    tjsonGetInt32ValueFromDouble(vnode, "dropped", pCfg->dropped, code);
    if (code != 0) goto _OVER;
    tjsonGetInt32ValueFromDouble(vnode, "vgVersion", pCfg->vgVersion, code);
    if (code != 0) goto _OVER;
    tjsonGetInt32ValueFromDouble(vnode, "diskPrimary", pCfg->diskPrimary, code);
    if (code != 0) goto _OVER;
    tjsonGetInt32ValueFromDouble(vnode, "toVgId", pCfg->toVgId, code);
    if (code != 0) goto _OVER;

    snprintf(pCfg->path, sizeof(pCfg->path), "%s%svnode%d", pMgmt->path, TD_DIRSEP, pCfg->vgId);
  }

  code = 0;
  *ppCfgs = pCfgs;
  *numOfVnodes = vnodesNum;

_OVER:
  if (*ppCfgs == NULL) taosMemoryFree(pCfgs);
  return code;
}

int32_t vmGetVnodeListFromFile(SVnodeMgmt *pMgmt, SWrapperCfg **ppCfgs, int32_t *numOfVnodes) {
  int32_t      code = -1;
  TdFilePtr    pFile = NULL;
  char        *pData = NULL;
  SJson       *pJson = NULL;
  char         file[PATH_MAX] = {0};
  SWrapperCfg *pCfgs = NULL;
  snprintf(file, sizeof(file), "%s%svnodes.json", pMgmt->path, TD_DIRSEP);

  if (taosStatFile(file, NULL, NULL, NULL) < 0) {
    code = terrno;
    dInfo("vnode file:%s not exist, reason:%s", file, tstrerror(code));
    code = 0;
    return code;
  }

  pFile = taosOpenFile(file, TD_FILE_READ);
  if (pFile == NULL) {
    code = terrno;
    dError("failed to open vnode file:%s since %s", file, tstrerror(code));
    goto _OVER;
  }

  int64_t size = 0;
  code = taosFStatFile(pFile, &size, NULL);
  if (code != 0) {
    dError("failed to fstat mnode file:%s since %s", file, tstrerror(code));
    goto _OVER;
  }

  pData = taosMemoryMalloc(size + 1);
  if (pData == NULL) {
    code = terrno;
    goto _OVER;
  }

  if (taosReadFile(pFile, pData, size) != size) {
    code = terrno;
    dError("failed to read vnode file:%s since %s", file, tstrerror(code));
    goto _OVER;
  }

  pData[size] = '\0';

  pJson = tjsonParse(pData);
  if (pJson == NULL) {
    code = TSDB_CODE_INVALID_JSON_FORMAT;
    goto _OVER;
  }

  if (vmDecodeVnodeList(pJson, pMgmt, ppCfgs, numOfVnodes) < 0) {
    code = TSDB_CODE_INVALID_JSON_FORMAT;
    goto _OVER;
  }

  code = 0;
  dInfo("succceed to read vnode file %s", file);

_OVER:
  if (pData != NULL) taosMemoryFree(pData);
  if (pJson != NULL) cJSON_Delete(pJson);
  if (pFile != NULL) taosCloseFile(&pFile);

  if (code != 0) {
    dError("failed to read vnode file:%s since %s", file, tstrerror(code));
  }
  return code;
}

static int32_t vmEncodeVnodeList(SJson *pJson, SVnodeObj **ppVnodes, int32_t numOfVnodes) {
  int32_t code = 0;
  SJson  *vnodes = tjsonCreateArray();
  if (vnodes == NULL) {
    return terrno;
  }
  if ((code = tjsonAddItemToObject(pJson, "vnodes", vnodes)) < 0) {
    tjsonDelete(vnodes);
    return code;
  };

  for (int32_t i = 0; i < numOfVnodes; ++i) {
    SVnodeObj *pVnode = ppVnodes[i];
    if (pVnode == NULL) continue;

    SJson *vnode = tjsonCreateObject();
    if (vnode == NULL) return terrno;
    if ((code = tjsonAddDoubleToObject(vnode, "vgId", pVnode->vgId)) < 0) return code;
    if ((code = tjsonAddDoubleToObject(vnode, "dropped", pVnode->dropped)) < 0) return code;
    if ((code = tjsonAddDoubleToObject(vnode, "vgVersion", pVnode->vgVersion)) < 0) return code;
    if ((code = tjsonAddDoubleToObject(vnode, "diskPrimary", pVnode->diskPrimary)) < 0) return code;
    if (pVnode->toVgId) {
      if ((code = tjsonAddDoubleToObject(vnode, "toVgId", pVnode->toVgId)) < 0) return code;
    }
    if ((code = tjsonAddItemToArray(vnodes, vnode)) < 0) return code;
  }

  return 0;
}

int32_t vmWriteVnodeListToFile(SVnodeMgmt *pMgmt) {
  int32_t     code = -1;
  char       *buffer = NULL;
  SJson      *pJson = NULL;
  TdFilePtr   pFile = NULL;
  SVnodeObj **ppVnodes = NULL;
  char        file[PATH_MAX] = {0};
  char        realfile[PATH_MAX] = {0};
  int32_t     lino = 0;
  int32_t     ret = -1;

  int32_t nBytes = snprintf(file, sizeof(file), "%s%svnodes_tmp.json", pMgmt->path, TD_DIRSEP);
  if (nBytes <= 0 || nBytes >= sizeof(file)) {
    return TSDB_CODE_OUT_OF_RANGE;
  }

  nBytes = snprintf(realfile, sizeof(realfile), "%s%svnodes.json", pMgmt->path, TD_DIRSEP);
  if (nBytes <= 0 || nBytes >= sizeof(realfile)) {
    return TSDB_CODE_OUT_OF_RANGE;
  }

  int32_t numOfVnodes = 0;
  TAOS_CHECK_GOTO(vmGetAllVnodeListFromHash(pMgmt, &numOfVnodes, &ppVnodes), &lino, _OVER);

  // terrno = TSDB_CODE_OUT_OF_MEMORY;
  pJson = tjsonCreateObject();
  if (pJson == NULL) {
    code = terrno;
    goto _OVER;
  }
  TAOS_CHECK_GOTO(vmEncodeVnodeList(pJson, ppVnodes, numOfVnodes), &lino, _OVER);

  buffer = tjsonToString(pJson);
  if (buffer == NULL) {
    code = TSDB_CODE_INVALID_JSON_FORMAT;
    lino = __LINE__;
    goto _OVER;
  }

  code = taosThreadMutexLock(&pMgmt->fileLock);
  if (code != 0) {
    lino = __LINE__;
    goto _OVER;
  }

  pFile = taosOpenFile(file, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_TRUNC | TD_FILE_WRITE_THROUGH);
  if (pFile == NULL) {
    code = terrno;
    lino = __LINE__;
    goto _OVER1;
  }

  int32_t len = strlen(buffer);
  if (taosWriteFile(pFile, buffer, len) <= 0) {
    code = terrno;
    lino = __LINE__;
    goto _OVER1;
  }
  if (taosFsyncFile(pFile) < 0) {
    code = TAOS_SYSTEM_ERROR(ERRNO);
    lino = __LINE__;
    goto _OVER1;
  }

  code = taosCloseFile(&pFile);
  if (code != 0) {
    code = TAOS_SYSTEM_ERROR(ERRNO);
    lino = __LINE__;
    goto _OVER1;
  }
  TAOS_CHECK_GOTO(taosRenameFile(file, realfile), &lino, _OVER1);

  dInfo("succeed to write vnodes file:%s, vnodes:%d", realfile, numOfVnodes);

_OVER1:
  ret = taosThreadMutexUnlock(&pMgmt->fileLock);
  if (ret != 0) {
    dError("failed to unlock since %s", tstrerror(ret));
  }

_OVER:
  if (pJson != NULL) tjsonDelete(pJson);
  if (buffer != NULL) taosMemoryFree(buffer);
  if (pFile != NULL) taosCloseFile(&pFile);
  if (ppVnodes != NULL) {
    for (int32_t i = 0; i < numOfVnodes; ++i) {
      SVnodeObj *pVnode = ppVnodes[i];
      if (pVnode != NULL) {
        vmReleaseVnode(pMgmt, pVnode);
      }
    }
    taosMemoryFree(ppVnodes);
  }

  if (code != 0) {
    dError("failed to write vnodes file:%s at line:%d since %s, vnodes:%d", realfile, lino, tstrerror(code),
           numOfVnodes);
  }
  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	#include "tjson.h"
18	#include "vmInt.h"
19
20	#define MAX_CONTENT_LEN 2 * 1024 * 1024
21
22	int32_t vmGetAllVnodeListFromHash(SVnodeMgmt pMgmt, int32_t numOfVnodes, SVnodeObj ***ppVnodes) {	17,517✔
23	(void)taosThreadRwlockRdlock(&pMgmt->hashLock);	17,517✔
24
25	int32_t num = 0;	17,517✔
26	int32_t size = taosHashGetSize(pMgmt->runngingHash);	17,517✔
27	int32_t closedSize = taosHashGetSize(pMgmt->closedHash);	17,517✔
28	size += closedSize;	17,517✔
29	SVnodeObj *pVnodes = taosMemoryCalloc(size, sizeof(SVnodeObj ));	17,517!
30	if (pVnodes == NULL) {	17,517!
31	(void)taosThreadRwlockUnlock(&pMgmt->hashLock);	×
32	return terrno;	×
33	}
34
35	void *pIter = taosHashIterate(pMgmt->runngingHash, NULL);	17,517✔
36	while (pIter) {	82,991✔
37	SVnodeObj **ppVnode = pIter;	65,474✔
38	SVnodeObj pVnode = ppVnode;	65,474✔
39	if (pVnode && num < size) {	130,948!
40	int32_t refCount = atomic_add_fetch_32(&pVnode->refCount, 1);	65,474✔
41	dTrace("vgId:%d,acquire vnode, vnode:%p, ref:%d", pVnode->vgId, pVnode, refCount);	65,474✔
42	pVnodes[num++] = (*ppVnode);	65,474✔
43	pIter = taosHashIterate(pMgmt->runngingHash, pIter);	65,474✔
44	} else {
45	taosHashCancelIterate(pMgmt->runngingHash, pIter);	×
46	}
47	}
48
49	pIter = taosHashIterate(pMgmt->closedHash, NULL);	17,517✔
50	while (pIter) {	17,519✔
51	SVnodeObj **ppVnode = pIter;	2✔
52	SVnodeObj pVnode = ppVnode;	2✔
53	if (pVnode && num < size) {	4!
54	int32_t refCount = atomic_add_fetch_32(&pVnode->refCount, 1);	2✔
55	dTrace("vgId:%d, acquire vnode, vnode:%p, ref:%d", pVnode->vgId, pVnode, refCount);	2!
56	pVnodes[num++] = (*ppVnode);	2✔
57	pIter = taosHashIterate(pMgmt->closedHash, pIter);	2✔
58	} else {
59	taosHashCancelIterate(pMgmt->closedHash, pIter);	×
60	}
61	}
62
63	(void)taosThreadRwlockUnlock(&pMgmt->hashLock);	17,517✔
64	*numOfVnodes = num;	17,517✔
65	*ppVnodes = pVnodes;	17,517✔
66
67	return 0;	17,517✔
68	}
69
70	int32_t vmGetAllVnodeListFromHashWithCreating(SVnodeMgmt pMgmt, int32_t numOfVnodes, SVnodeObj ***ppVnodes) {	9,927✔
71	(void)taosThreadRwlockRdlock(&pMgmt->hashLock);	9,927✔
72
73	int32_t num = 0;	9,927✔
74	int32_t size = taosHashGetSize(pMgmt->runngingHash);	9,927✔
75	int32_t creatingSize = taosHashGetSize(pMgmt->creatingHash);	9,927✔
76	size += creatingSize;	9,927✔
77	SVnodeObj *pVnodes = taosMemoryCalloc(size, sizeof(SVnodeObj ));	9,927!
78	if (pVnodes == NULL) {	9,927!
79	(void)taosThreadRwlockUnlock(&pMgmt->hashLock);	×
80	return terrno;	×
81	}
82
83	void *pIter = taosHashIterate(pMgmt->runngingHash, NULL);	9,927✔
84	while (pIter) {	29,087✔
85	SVnodeObj **ppVnode = pIter;	19,160✔
86	SVnodeObj pVnode = ppVnode;	19,160✔
87	if (pVnode && num < size) {	38,320!
88	int32_t refCount = atomic_add_fetch_32(&pVnode->refCount, 1);	19,160✔
89	dTrace("vgId:%d,acquire vnode, vnode:%p, ref:%d", pVnode->vgId, pVnode, refCount);	19,160✔
90	pVnodes[num++] = (*ppVnode);	19,160✔
91	pIter = taosHashIterate(pMgmt->runngingHash, pIter);	19,160✔
92	} else {
93	taosHashCancelIterate(pMgmt->runngingHash, pIter);	×
94	}
95	}
96
97	pIter = taosHashIterate(pMgmt->creatingHash, NULL);	9,927✔
98	while (pIter) {	19,063✔
99	SVnodeObj **ppVnode = pIter;	9,136✔
100	SVnodeObj pVnode = ppVnode;	9,136✔
101	if (pVnode && num < size) {	18,272!
102	int32_t refCount = atomic_add_fetch_32(&pVnode->refCount, 1);	9,136✔
103	dTrace("vgId:%d, acquire vnode, vnode:%p, ref:%d", pVnode->vgId, pVnode, refCount);	9,136✔
104	pVnodes[num++] = (*ppVnode);	9,136✔
105	pIter = taosHashIterate(pMgmt->creatingHash, pIter);	9,136✔
106	} else {
107	taosHashCancelIterate(pMgmt->creatingHash, pIter);	×
108	}
109	}
110	(void)taosThreadRwlockUnlock(&pMgmt->hashLock);	9,927✔
111
112	*numOfVnodes = num;	9,927✔
113	*ppVnodes = pVnodes;	9,927✔
114
115	return 0;	9,927✔
116	}
117
118	int32_t vmGetVnodeListFromHash(SVnodeMgmt pMgmt, int32_t numOfVnodes, SVnodeObj ***ppVnodes) {	7,498✔
119	(void)taosThreadRwlockRdlock(&pMgmt->hashLock);	7,498✔
120
121	int32_t num = 0;	7,498✔
122	int32_t size = taosHashGetSize(pMgmt->runngingHash);	7,498✔
123	SVnodeObj *pVnodes = taosMemoryCalloc(size, sizeof(SVnodeObj ));	7,498!
124	if (pVnodes == NULL) {	7,498!
125	(void)taosThreadRwlockUnlock(&pMgmt->hashLock);	×
126	return terrno;	×
127	}
128
129	void *pIter = taosHashIterate(pMgmt->runngingHash, NULL);	7,498✔
130	while (pIter) {	21,935✔
131	SVnodeObj **ppVnode = pIter;	14,437✔
132	SVnodeObj pVnode = ppVnode;	14,437✔
133	if (pVnode && num < size) {	28,874!
134	int32_t refCount = atomic_add_fetch_32(&pVnode->refCount, 1);	14,437✔
135	dTrace("vgId:%d, acquire vnode, vnode:%p, ref:%d", pVnode->vgId, pVnode, refCount);	14,437✔
136	pVnodes[num++] = (*ppVnode);	14,437✔
137	pIter = taosHashIterate(pMgmt->runngingHash, pIter);	14,437✔
138	} else {
139	taosHashCancelIterate(pMgmt->runngingHash, pIter);	×
140	}
141	}
142
143	(void)taosThreadRwlockUnlock(&pMgmt->hashLock);	7,498✔
144	*numOfVnodes = num;	7,498✔
145	*ppVnodes = pVnodes;	7,498✔
146
147	return 0;	7,498✔
148	}
149
150	static int32_t vmDecodeVnodeList(SJson pJson, SVnodeMgmt pMgmt, SWrapperCfg *ppCfgs, int32_t numOfVnodes) {	422✔
151	int32_t code = -1;	422✔
152	SWrapperCfg *pCfgs = NULL;	422✔
153	*ppCfgs = NULL;	422✔
154
155	SJson *vnodes = tjsonGetObjectItem(pJson, "vnodes");	422✔
156	if (vnodes == NULL) return TSDB_CODE_INVALID_JSON_FORMAT;	422!
157
158	int32_t vnodesNum = cJSON_GetArraySize(vnodes);	422✔
159	if (vnodesNum > 0) {	422✔
160	pCfgs = taosMemoryCalloc(vnodesNum, sizeof(SWrapperCfg));	419!
161	if (pCfgs == NULL) return terrno;	419!
162	}
163
164	for (int32_t i = 0; i < vnodesNum; ++i) {	1,561✔
165	SJson *vnode = tjsonGetArrayItem(vnodes, i);	1,139✔
166	if (vnode == NULL) {	1,139!
167	code = TSDB_CODE_INVALID_JSON_FORMAT;	×
168	goto _OVER;	×
169	}
170
171	SWrapperCfg *pCfg = &pCfgs[i];	1,139✔
172	tjsonGetInt32ValueFromDouble(vnode, "vgId", pCfg->vgId, code);	1,139✔
173	if (code != 0) goto _OVER;	1,139!
174	tjsonGetInt32ValueFromDouble(vnode, "dropped", pCfg->dropped, code);	1,139✔
175	if (code != 0) goto _OVER;	1,139!
176	tjsonGetInt32ValueFromDouble(vnode, "vgVersion", pCfg->vgVersion, code);	1,139✔
177	if (code != 0) goto _OVER;	1,139!
178	tjsonGetInt32ValueFromDouble(vnode, "diskPrimary", pCfg->diskPrimary, code);	1,139✔
179	if (code != 0) goto _OVER;	1,139!
180	tjsonGetInt32ValueFromDouble(vnode, "toVgId", pCfg->toVgId, code);	1,139✔
181	if (code != 0) goto _OVER;	1,139!
182
183	snprintf(pCfg->path, sizeof(pCfg->path), "%s%svnode%d", pMgmt->path, TD_DIRSEP, pCfg->vgId);	1,139✔
184	}
185
186	code = 0;	422✔
187	*ppCfgs = pCfgs;	422✔
188	*numOfVnodes = vnodesNum;	422✔
189
190	_OVER:	422✔
191	if (*ppCfgs == NULL) taosMemoryFree(pCfgs);	422!
192	return code;	422✔
193	}
194
195	int32_t vmGetVnodeListFromFile(SVnodeMgmt pMgmt, SWrapperCfg ppCfgs, int32_t numOfVnodes) {	2,606✔
196	int32_t code = -1;	2,606✔
197	TdFilePtr pFile = NULL;	2,606✔
198	char *pData = NULL;	2,606✔
199	SJson *pJson = NULL;	2,606✔
200	char file[PATH_MAX] = {0};	2,606✔
201	SWrapperCfg *pCfgs = NULL;	2,606✔
202	snprintf(file, sizeof(file), "%s%svnodes.json", pMgmt->path, TD_DIRSEP);	2,606✔
203
204	if (taosStatFile(file, NULL, NULL, NULL) < 0) {	2,606✔
205	code = terrno;	2,184✔
206	dInfo("vnode file:%s not exist, reason:%s", file, tstrerror(code));	2,184!
207	code = 0;	2,184✔
208	return code;	2,184✔
209	}
210
211	pFile = taosOpenFile(file, TD_FILE_READ);	422✔
212	if (pFile == NULL) {	422!
213	code = terrno;	×
214	dError("failed to open vnode file:%s since %s", file, tstrerror(code));	×
215	goto _OVER;	×
216	}
217
218	int64_t size = 0;	422✔
219	code = taosFStatFile(pFile, &size, NULL);	422✔
220	if (code != 0) {	422!
221	dError("failed to fstat mnode file:%s since %s", file, tstrerror(code));	×
222	goto _OVER;	×
223	}
224
225	pData = taosMemoryMalloc(size + 1);	422!
226	if (pData == NULL) {	422!
227	code = terrno;	×
228	goto _OVER;	×
229	}
230
231	if (taosReadFile(pFile, pData, size) != size) {	422!
232	code = terrno;	×
233	dError("failed to read vnode file:%s since %s", file, tstrerror(code));	×
234	goto _OVER;	×
235	}
236
237	pData[size] = '\0';	422✔
238
239	pJson = tjsonParse(pData);	422✔
240	if (pJson == NULL) {	422!
241	code = TSDB_CODE_INVALID_JSON_FORMAT;	×
242	goto _OVER;	×
243	}
244
245	if (vmDecodeVnodeList(pJson, pMgmt, ppCfgs, numOfVnodes) < 0) {	422!
246	code = TSDB_CODE_INVALID_JSON_FORMAT;	×
247	goto _OVER;	×
248	}
249
250	code = 0;	422✔
251	dInfo("succceed to read vnode file %s", file);	422!
252
253	_OVER:	×
254	if (pData != NULL) taosMemoryFree(pData);	422!
255	if (pJson != NULL) cJSON_Delete(pJson);	422!
256	if (pFile != NULL) taosCloseFile(&pFile);	422!
257
258	if (code != 0) {	422!
259	dError("failed to read vnode file:%s since %s", file, tstrerror(code));	×
260	}
261	return code;	422✔
262	}
263
264	static int32_t vmEncodeVnodeList(SJson pJson, SVnodeObj *ppVnodes, int32_t numOfVnodes) {	17,517✔
265	int32_t code = 0;	17,517✔
266	SJson *vnodes = tjsonCreateArray();	17,517✔
267	if (vnodes == NULL) {	17,516!
268	return terrno;	×
269	}
270	if ((code = tjsonAddItemToObject(pJson, "vnodes", vnodes)) < 0) {	17,516!
271	tjsonDelete(vnodes);	×
272	return code;	×
273	};
274
275	for (int32_t i = 0; i < numOfVnodes; ++i) {	82,993✔
276	SVnodeObj *pVnode = ppVnodes[i];	65,474✔
277	if (pVnode == NULL) continue;	65,474!
278
279	SJson *vnode = tjsonCreateObject();	65,474✔
280	if (vnode == NULL) return terrno;	65,475!
281	if ((code = tjsonAddDoubleToObject(vnode, "vgId", pVnode->vgId)) < 0) return code;	65,475!
282	if ((code = tjsonAddDoubleToObject(vnode, "dropped", pVnode->dropped)) < 0) return code;	65,474!
283	if ((code = tjsonAddDoubleToObject(vnode, "vgVersion", pVnode->vgVersion)) < 0) return code;	65,476!
284	if ((code = tjsonAddDoubleToObject(vnode, "diskPrimary", pVnode->diskPrimary)) < 0) return code;	65,476!
285	if (pVnode->toVgId) {	65,476✔
286	if ((code = tjsonAddDoubleToObject(vnode, "toVgId", pVnode->toVgId)) < 0) return code;	30!
287	}
288	if ((code = tjsonAddItemToArray(vnodes, vnode)) < 0) return code;	65,476!
289	}
290
291	return 0;	17,519✔
292	}
293
294	int32_t vmWriteVnodeListToFile(SVnodeMgmt *pMgmt) {	17,517✔
295	int32_t code = -1;	17,517✔
296	char *buffer = NULL;	17,517✔
297	SJson *pJson = NULL;	17,517✔
298	TdFilePtr pFile = NULL;	17,517✔
299	SVnodeObj **ppVnodes = NULL;	17,517✔
300	char file[PATH_MAX] = {0};	17,517✔
301	char realfile[PATH_MAX] = {0};	17,517✔
302	int32_t lino = 0;	17,517✔
303	int32_t ret = -1;	17,517✔
304
305	int32_t nBytes = snprintf(file, sizeof(file), "%s%svnodes_tmp.json", pMgmt->path, TD_DIRSEP);	17,517✔
306	if (nBytes <= 0 \|\| nBytes >= sizeof(file)) {	17,517!
307	return TSDB_CODE_OUT_OF_RANGE;	×
308	}
309
310	nBytes = snprintf(realfile, sizeof(realfile), "%s%svnodes.json", pMgmt->path, TD_DIRSEP);	17,517✔
311	if (nBytes <= 0 \|\| nBytes >= sizeof(realfile)) {	17,517!
312	return TSDB_CODE_OUT_OF_RANGE;	×
313	}
314
315	int32_t numOfVnodes = 0;	17,517✔
316	TAOS_CHECK_GOTO(vmGetAllVnodeListFromHash(pMgmt, &numOfVnodes, &ppVnodes), &lino, _OVER);	17,517!
317
318	// terrno = TSDB_CODE_OUT_OF_MEMORY;
319	pJson = tjsonCreateObject();	17,517✔
320	if (pJson == NULL) {	17,517!
321	code = terrno;	×
322	goto _OVER;	×
323	}
324	TAOS_CHECK_GOTO(vmEncodeVnodeList(pJson, ppVnodes, numOfVnodes), &lino, _OVER);	17,517!
325
326	buffer = tjsonToString(pJson);	17,517✔
327	if (buffer == NULL) {	17,516!
328	code = TSDB_CODE_INVALID_JSON_FORMAT;	×
329	lino = __LINE__;	×
330	goto _OVER;	×
331	}
332
333	code = taosThreadMutexLock(&pMgmt->fileLock);	17,516✔
334	if (code != 0) {	17,517!
335	lino = __LINE__;	×
336	goto _OVER;	×
337	}
338
339	pFile = taosOpenFile(file, TD_FILE_CREATE \| TD_FILE_WRITE \| TD_FILE_TRUNC \| TD_FILE_WRITE_THROUGH);	17,517✔
340	if (pFile == NULL) {	17,517!
341	code = terrno;	×
342	lino = __LINE__;	×
343	goto _OVER1;	×
344	}
345
346	int32_t len = strlen(buffer);	17,517✔
347	if (taosWriteFile(pFile, buffer, len) <= 0) {	17,517!
348	code = terrno;	×
349	lino = __LINE__;	×
350	goto _OVER1;	×
351	}
352	if (taosFsyncFile(pFile) < 0) {	17,517!
353	code = TAOS_SYSTEM_ERROR(ERRNO);	×
354	lino = __LINE__;	×
355	goto _OVER1;	×
356	}
357
358	code = taosCloseFile(&pFile);	17,517✔
359	if (code != 0) {	17,517!
360	code = TAOS_SYSTEM_ERROR(ERRNO);	×
361	lino = __LINE__;	×
362	goto _OVER1;	×
363	}
364	TAOS_CHECK_GOTO(taosRenameFile(file, realfile), &lino, _OVER1);	17,517!
365
366	dInfo("succeed to write vnodes file:%s, vnodes:%d", realfile, numOfVnodes);	17,517!
367
368	_OVER1:	×
369	ret = taosThreadMutexUnlock(&pMgmt->fileLock);	17,517✔
370	if (ret != 0) {	17,517!
371	dError("failed to unlock since %s", tstrerror(ret));	×
372	}
373
374	_OVER:	17,517✔
375	if (pJson != NULL) tjsonDelete(pJson);	17,517!
376	if (buffer != NULL) taosMemoryFree(buffer);	17,517!
377	if (pFile != NULL) taosCloseFile(&pFile);	17,517!
378	if (ppVnodes != NULL) {	17,517!
379	for (int32_t i = 0; i < numOfVnodes; ++i) {	82,993✔
380	SVnodeObj *pVnode = ppVnodes[i];	65,476✔
381	if (pVnode != NULL) {	65,476!
382	vmReleaseVnode(pMgmt, pVnode);	65,476✔
383	}
384	}
385	taosMemoryFree(ppVnodes);	17,517!
386	}
387
388	if (code != 0) {	17,517!
389	dError("failed to write vnodes file:%s at line:%d since %s, vnodes:%d", realfile, lino, tstrerror(code),	×
390	numOfVnodes);
391	}
392	return code;	17,517✔
393	}

taosdata / TDengine / #4361

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