#3562

Committed 20 Dec 2024 09:57AM UTC coverage: 26.655% (-32.2%) from 58.812%

Build # #3562

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

Merge pull request #29229 from taosdata/enh/TS-5749-3.0

enh: seperate tsdb async tasks to different thread pools

Run Details

21498 of 109421 branches covered (19.65%)

Branch coverage included in aggregate %.

66 of 96 new or added lines in 7 files covered. (68.75%)

39441 existing lines in 157 files now uncovered.

35007 of 102566 relevant lines covered (34.13%)

53922.97 hits per line

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

9.94

/source/dnode/mnode/impl/src/mndTelem.c

/*
 * Copyright (c) 2020 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 "mndTelem.h"
#include "mndCluster.h"
#include "mndSync.h"
#include "thttp.h"
#include "tjson.h"

typedef struct {
  int64_t numOfDnode;
  int64_t numOfMnode;
  int64_t numOfVgroup;
  int64_t numOfDatabase;
  int64_t numOfSuperTable;
  int64_t numOfChildTable;
  int64_t numOfNormalTable;
  int64_t numOfColumn;
  int64_t totalPoints;
  int64_t totalStorage;
  int64_t compStorage;
} SMnodeStat;

static void mndGetStat(SMnode* pMnode, SMnodeStat* pStat) {
  memset(pStat, 0, sizeof(SMnodeStat));

  SSdb* pSdb = pMnode->pSdb;
  pStat->numOfDnode = sdbGetSize(pSdb, SDB_DNODE);
  pStat->numOfMnode = sdbGetSize(pSdb, SDB_MNODE);
  pStat->numOfVgroup = sdbGetSize(pSdb, SDB_VGROUP);
  pStat->numOfDatabase = sdbGetSize(pSdb, SDB_DB);
  pStat->numOfSuperTable = sdbGetSize(pSdb, SDB_STB);

  void* pIter = NULL;
  while (1) {
    SVgObj* pVgroup = NULL;
    pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void**)&pVgroup);
    if (pIter == NULL) break;

    pStat->numOfChildTable += pVgroup->numOfTables;
    pStat->numOfColumn += pVgroup->numOfTimeSeries;
    pStat->totalPoints += pVgroup->pointsWritten;
    pStat->totalStorage += pVgroup->totalStorage;
    pStat->compStorage += pVgroup->compStorage;

    sdbRelease(pSdb, pVgroup);
  }

  pStat->numOfChildTable = 100;
  pStat->numOfColumn = 200;
  pStat->totalPoints = 300;
  pStat->totalStorage = 400;
  pStat->compStorage = 500;
}

static void mndBuildRuntimeInfo(SMnode* pMnode, SJson* pJson) {
  SMnodeStat mstat = {0};
  int32_t    code = 0;
  int32_t    lino = 0;
  mndGetStat(pMnode, &mstat);

  TAOS_CHECK_GOTO(tjsonAddDoubleToObject(pJson, "numOfDnode", mstat.numOfDnode), &lino, _OVER);
  TAOS_CHECK_GOTO(tjsonAddDoubleToObject(pJson, "numOfMnode", mstat.numOfMnode), &lino, _OVER);
  TAOS_CHECK_GOTO(tjsonAddDoubleToObject(pJson, "numOfVgroup", mstat.numOfVgroup), &lino, _OVER);
  TAOS_CHECK_GOTO(tjsonAddDoubleToObject(pJson, "numOfDatabase", mstat.numOfDatabase), &lino, _OVER);
  TAOS_CHECK_GOTO(tjsonAddDoubleToObject(pJson, "numOfSuperTable", mstat.numOfSuperTable), &lino, _OVER);
  TAOS_CHECK_GOTO(tjsonAddDoubleToObject(pJson, "numOfChildTable", mstat.numOfChildTable), &lino, _OVER);
  TAOS_CHECK_GOTO(tjsonAddDoubleToObject(pJson, "numOfColumn", mstat.numOfColumn), &lino, _OVER);
  TAOS_CHECK_GOTO(tjsonAddDoubleToObject(pJson, "numOfPoint", mstat.totalPoints), &lino, _OVER);
  TAOS_CHECK_GOTO(tjsonAddDoubleToObject(pJson, "totalStorage", mstat.totalStorage), &lino, _OVER);
  TAOS_CHECK_GOTO(tjsonAddDoubleToObject(pJson, "compStorage", mstat.compStorage), &lino, _OVER);
_OVER:
  if (code != 0) mError("failed to mndBuildRuntimeInfo at line:%d since %s", lino, tstrerror(code));
}

static char* mndBuildTelemetryReport(SMnode* pMnode) {
  char        tmp[4096] = {0};
  STelemMgmt* pMgmt = &pMnode->telemMgmt;
  int32_t     code = 0;
  int32_t     lino = 0;

  SJson* pJson = tjsonCreateObject();
  if (pJson == NULL) return NULL;

  char clusterName[64] = {0};
  if ((terrno = mndGetClusterName(pMnode, clusterName, sizeof(clusterName))) != 0) return NULL;
  TAOS_CHECK_GOTO(tjsonAddStringToObject(pJson, "instanceId", clusterName), &lino, _OVER);
  TAOS_CHECK_GOTO(tjsonAddDoubleToObject(pJson, "reportVersion", 1), &lino, _OVER);

  if (taosGetOsReleaseName(tmp, NULL, NULL, sizeof(tmp)) == 0) {
    TAOS_CHECK_GOTO(tjsonAddStringToObject(pJson, "os", tmp), &lino, _OVER);
  }

  float numOfCores = 0;
  if (taosGetCpuInfo(tmp, sizeof(tmp), &numOfCores) == 0) {
    TAOS_CHECK_GOTO(tjsonAddStringToObject(pJson, "cpuModel", tmp), &lino, _OVER);
    TAOS_CHECK_GOTO(tjsonAddDoubleToObject(pJson, "numOfCpu", numOfCores), &lino, _OVER);
  } else {
    TAOS_CHECK_GOTO(tjsonAddDoubleToObject(pJson, "numOfCpu", tsNumOfCores), &lino, _OVER);
  }

  snprintf(tmp, sizeof(tmp), "%" PRId64 " kB", tsTotalMemoryKB);
  TAOS_CHECK_GOTO(tjsonAddStringToObject(pJson, "memory", tmp), &lino, _OVER);

  TAOS_CHECK_GOTO(tjsonAddStringToObject(pJson, "version", td_version), &lino, _OVER);
  TAOS_CHECK_GOTO(tjsonAddStringToObject(pJson, "buildInfo", td_buildinfo), &lino, _OVER);
  TAOS_CHECK_GOTO(tjsonAddStringToObject(pJson, "gitInfo", td_gitinfo), &lino, _OVER);
  TAOS_CHECK_GOTO(tjsonAddStringToObject(pJson, "email", pMgmt->email), &lino, _OVER);

  mndBuildRuntimeInfo(pMnode, pJson);

_OVER:
  if (code != 0) {
    mError("failed to build telemetry report at lino:%d, since %s", lino, tstrerror(code));
  }
  char* pCont = tjsonToString(pJson);
  tjsonDelete(pJson);
  return pCont;
}

static int32_t mndProcessTelemTimer(SRpcMsg* pReq) {
  int32_t     code = 0;
  int32_t     line = 0;
  SMnode*     pMnode = pReq->info.node;
  STelemMgmt* pMgmt = &pMnode->telemMgmt;
  if (!tsEnableTelem) return 0;

  (void)taosThreadMutexLock(&pMgmt->lock);
  char* pCont = mndBuildTelemetryReport(pMnode);
  (void)taosThreadMutexUnlock(&pMgmt->lock);

  if (pCont == NULL) {
    return 0;
  }
  code = taosSendTelemReport(&pMgmt->addrMgt, tsTelemUri, tsTelemPort, pCont, strlen(pCont), HTTP_FLAT);
  taosMemoryFree(pCont);
  return code;
_end:
  if (code != 0) {
    mError("%s failed to send at line %d since %s", __func__, line, tstrerror(code));
  }
  taosMemoryFree(pCont);
  return code;
}

int32_t mndInitTelem(SMnode* pMnode) {
  int32_t     code = 0;
  STelemMgmt* pMgmt = &pMnode->telemMgmt;

  (void)taosThreadMutexInit(&pMgmt->lock, NULL);
  if ((code = taosGetEmail(pMgmt->email, sizeof(pMgmt->email))) != 0)
    mWarn("failed to get email since %s", tstrerror(code));
  code = taosTelemetryMgtInit(&pMgmt->addrMgt, tsTelemServer);
  if (code != 0) {
    mError("failed to init telemetry management since %s", tstrerror(code));
    return code;
  }
  mndSetMsgHandle(pMnode, TDMT_MND_TELEM_TIMER, mndProcessTelemTimer);

  return 0;
}

void mndCleanupTelem(SMnode* pMnode) {
  STelemMgmt* pMgmt = &pMnode->telemMgmt;
  taosTelemetryDestroy(&pMgmt->addrMgt);
  (void)taosThreadMutexDestroy(&pMgmt->lock);
}

1	/*
2	* Copyright (c) 2020 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 "mndTelem.h"
18	#include "mndCluster.h"
19	#include "mndSync.h"
20	#include "thttp.h"
21	#include "tjson.h"
22
23	typedef struct {
24	int64_t numOfDnode;
25	int64_t numOfMnode;
26	int64_t numOfVgroup;
27	int64_t numOfDatabase;
28	int64_t numOfSuperTable;
29	int64_t numOfChildTable;
30	int64_t numOfNormalTable;
31	int64_t numOfColumn;
32	int64_t totalPoints;
33	int64_t totalStorage;
34	int64_t compStorage;
35	} SMnodeStat;
36
UNCOV 37	static void mndGetStat(SMnode* pMnode, SMnodeStat* pStat) {	×
UNCOV 38	memset(pStat, 0, sizeof(SMnodeStat));	×
39
UNCOV 40	SSdb* pSdb = pMnode->pSdb;	×
UNCOV 41	pStat->numOfDnode = sdbGetSize(pSdb, SDB_DNODE);	×
UNCOV 42	pStat->numOfMnode = sdbGetSize(pSdb, SDB_MNODE);	×
UNCOV 43	pStat->numOfVgroup = sdbGetSize(pSdb, SDB_VGROUP);	×
UNCOV 44	pStat->numOfDatabase = sdbGetSize(pSdb, SDB_DB);	×
UNCOV 45	pStat->numOfSuperTable = sdbGetSize(pSdb, SDB_STB);	×
46
UNCOV 47	void* pIter = NULL;	×
UNCOV 48	while (1) {	×
UNCOV 49	SVgObj* pVgroup = NULL;	×
UNCOV 50	pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void**)&pVgroup);	×
UNCOV 51	if (pIter == NULL) break;	×
52
UNCOV 53	pStat->numOfChildTable += pVgroup->numOfTables;	×
UNCOV 54	pStat->numOfColumn += pVgroup->numOfTimeSeries;	×
UNCOV 55	pStat->totalPoints += pVgroup->pointsWritten;	×
UNCOV 56	pStat->totalStorage += pVgroup->totalStorage;	×
UNCOV 57	pStat->compStorage += pVgroup->compStorage;	×
58
UNCOV 59	sdbRelease(pSdb, pVgroup);	×
60	}
61
UNCOV 62	pStat->numOfChildTable = 100;	×
UNCOV 63	pStat->numOfColumn = 200;	×
UNCOV 64	pStat->totalPoints = 300;	×
UNCOV 65	pStat->totalStorage = 400;	×
UNCOV 66	pStat->compStorage = 500;	×
UNCOV 67	}	×
68
UNCOV 69	static void mndBuildRuntimeInfo(SMnode* pMnode, SJson* pJson) {	×
UNCOV 70	SMnodeStat mstat = {0};	×
UNCOV 71	int32_t code = 0;	×
UNCOV 72	int32_t lino = 0;	×
UNCOV 73	mndGetStat(pMnode, &mstat);	×
74
UNCOV 75	TAOS_CHECK_GOTO(tjsonAddDoubleToObject(pJson, "numOfDnode", mstat.numOfDnode), &lino, _OVER);	×
UNCOV 76	TAOS_CHECK_GOTO(tjsonAddDoubleToObject(pJson, "numOfMnode", mstat.numOfMnode), &lino, _OVER);	×
UNCOV 77	TAOS_CHECK_GOTO(tjsonAddDoubleToObject(pJson, "numOfVgroup", mstat.numOfVgroup), &lino, _OVER);	×
UNCOV 78	TAOS_CHECK_GOTO(tjsonAddDoubleToObject(pJson, "numOfDatabase", mstat.numOfDatabase), &lino, _OVER);	×
UNCOV 79	TAOS_CHECK_GOTO(tjsonAddDoubleToObject(pJson, "numOfSuperTable", mstat.numOfSuperTable), &lino, _OVER);	×
UNCOV 80	TAOS_CHECK_GOTO(tjsonAddDoubleToObject(pJson, "numOfChildTable", mstat.numOfChildTable), &lino, _OVER);	×
UNCOV 81	TAOS_CHECK_GOTO(tjsonAddDoubleToObject(pJson, "numOfColumn", mstat.numOfColumn), &lino, _OVER);	×
UNCOV 82	TAOS_CHECK_GOTO(tjsonAddDoubleToObject(pJson, "numOfPoint", mstat.totalPoints), &lino, _OVER);	×
UNCOV 83	TAOS_CHECK_GOTO(tjsonAddDoubleToObject(pJson, "totalStorage", mstat.totalStorage), &lino, _OVER);	×
UNCOV 84	TAOS_CHECK_GOTO(tjsonAddDoubleToObject(pJson, "compStorage", mstat.compStorage), &lino, _OVER);	×
UNCOV 85	_OVER:	×
UNCOV 86	if (code != 0) mError("failed to mndBuildRuntimeInfo at line:%d since %s", lino, tstrerror(code));	×
UNCOV 87	}	×
88
UNCOV 89	static char* mndBuildTelemetryReport(SMnode* pMnode) {	×
UNCOV 90	char tmp[4096] = {0};	×
UNCOV 91	STelemMgmt* pMgmt = &pMnode->telemMgmt;	×
UNCOV 92	int32_t code = 0;	×
UNCOV 93	int32_t lino = 0;	×
94
UNCOV 95	SJson* pJson = tjsonCreateObject();	×
UNCOV 96	if (pJson == NULL) return NULL;	×
97
UNCOV 98	char clusterName[64] = {0};	×
UNCOV 99	if ((terrno = mndGetClusterName(pMnode, clusterName, sizeof(clusterName))) != 0) return NULL;	×
UNCOV 100	TAOS_CHECK_GOTO(tjsonAddStringToObject(pJson, "instanceId", clusterName), &lino, _OVER);	×
UNCOV 101	TAOS_CHECK_GOTO(tjsonAddDoubleToObject(pJson, "reportVersion", 1), &lino, _OVER);	×
102
UNCOV 103	if (taosGetOsReleaseName(tmp, NULL, NULL, sizeof(tmp)) == 0) {	×
UNCOV 104	TAOS_CHECK_GOTO(tjsonAddStringToObject(pJson, "os", tmp), &lino, _OVER);	×
105	}
106
UNCOV 107	float numOfCores = 0;	×
UNCOV 108	if (taosGetCpuInfo(tmp, sizeof(tmp), &numOfCores) == 0) {	×
UNCOV 109	TAOS_CHECK_GOTO(tjsonAddStringToObject(pJson, "cpuModel", tmp), &lino, _OVER);	×
UNCOV 110	TAOS_CHECK_GOTO(tjsonAddDoubleToObject(pJson, "numOfCpu", numOfCores), &lino, _OVER);	×
111	} else {
112	TAOS_CHECK_GOTO(tjsonAddDoubleToObject(pJson, "numOfCpu", tsNumOfCores), &lino, _OVER);	×
113	}
114
UNCOV 115	snprintf(tmp, sizeof(tmp), "%" PRId64 " kB", tsTotalMemoryKB);	×
UNCOV 116	TAOS_CHECK_GOTO(tjsonAddStringToObject(pJson, "memory", tmp), &lino, _OVER);	×
117
UNCOV 118	TAOS_CHECK_GOTO(tjsonAddStringToObject(pJson, "version", td_version), &lino, _OVER);	×
UNCOV 119	TAOS_CHECK_GOTO(tjsonAddStringToObject(pJson, "buildInfo", td_buildinfo), &lino, _OVER);	×
UNCOV 120	TAOS_CHECK_GOTO(tjsonAddStringToObject(pJson, "gitInfo", td_gitinfo), &lino, _OVER);	×
UNCOV 121	TAOS_CHECK_GOTO(tjsonAddStringToObject(pJson, "email", pMgmt->email), &lino, _OVER);	×
122
UNCOV 123	mndBuildRuntimeInfo(pMnode, pJson);	×
124
UNCOV 125	_OVER:	×
UNCOV 126	if (code != 0) {	×
127	mError("failed to build telemetry report at lino:%d, since %s", lino, tstrerror(code));	×
128	}
UNCOV 129	char* pCont = tjsonToString(pJson);	×
UNCOV 130	tjsonDelete(pJson);	×
UNCOV 131	return pCont;	×
132	}
133
UNCOV 134	static int32_t mndProcessTelemTimer(SRpcMsg* pReq) {	×
UNCOV 135	int32_t code = 0;	×
UNCOV 136	int32_t line = 0;	×
UNCOV 137	SMnode* pMnode = pReq->info.node;	×
UNCOV 138	STelemMgmt* pMgmt = &pMnode->telemMgmt;	×
UNCOV 139	if (!tsEnableTelem) return 0;	×
140
UNCOV 141	(void)taosThreadMutexLock(&pMgmt->lock);	×
UNCOV 142	char* pCont = mndBuildTelemetryReport(pMnode);	×
UNCOV 143	(void)taosThreadMutexUnlock(&pMgmt->lock);	×
144
UNCOV 145	if (pCont == NULL) {	×
146	return 0;	×
147	}
UNCOV 148	code = taosSendTelemReport(&pMgmt->addrMgt, tsTelemUri, tsTelemPort, pCont, strlen(pCont), HTTP_FLAT);	×
UNCOV 149	taosMemoryFree(pCont);	×
UNCOV 150	return code;	×
151	_end:
152	if (code != 0) {
153	mError("%s failed to send at line %d since %s", __func__, line, tstrerror(code));
154	}
155	taosMemoryFree(pCont);
156	return code;
157	}
158
159	int32_t mndInitTelem(SMnode* pMnode) {	13✔
160	int32_t code = 0;	13✔
161	STelemMgmt* pMgmt = &pMnode->telemMgmt;	13✔
162
163	(void)taosThreadMutexInit(&pMgmt->lock, NULL);	13✔
164	if ((code = taosGetEmail(pMgmt->email, sizeof(pMgmt->email))) != 0)	13!
165	mWarn("failed to get email since %s", tstrerror(code));	13!
166	code = taosTelemetryMgtInit(&pMgmt->addrMgt, tsTelemServer);	13✔
167	if (code != 0) {	13!
168	mError("failed to init telemetry management since %s", tstrerror(code));	×
169	return code;	×
170	}
171	mndSetMsgHandle(pMnode, TDMT_MND_TELEM_TIMER, mndProcessTelemTimer);	13✔
172
173	return 0;	13✔
174	}
175
176	void mndCleanupTelem(SMnode* pMnode) {	13✔
177	STelemMgmt* pMgmt = &pMnode->telemMgmt;	13✔
178	taosTelemetryDestroy(&pMgmt->addrMgt);	13✔
179	(void)taosThreadMutexDestroy(&pMgmt->lock);	13✔
180	}	13✔

taosdata / TDengine / #3562

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