#5071

Committed 17 May 2026 01:15AM UTC coverage: 63.054% (-10.3%) from 73.326%

Build # #5071

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

feat (TDgpt): Dynamic Model Synchronization Enhancements (#35344)

* refactor: do some internal refactor.

* fix: fix multiprocess sync issue.

* feat: add dynamic anomaly detection and forecasting services

* fix: log error message for undeploying model in exception handling

* Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* Potential fix for pull request finding

Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

* fix: handle undeploy when model exists only on disk

Agent-Logs-Url: https://github.com/taosdata/TDengine/sessions/286aafa0-c3ce-4c27-b803-2707571e9dc1

Co-authored-by: hjxilinx <8252296+hjxilinx@users.noreply.github.com>

* fix: guard dynamic registry concurrent access

Agent-Logs-Url: https://github.com/taosdata/TDengine/sessions/5e4db858-6458-40f4-ac28-d1b1b7f97c18

Co-authored-by: hjxilinx <8252296+hjxilinx@users.noreply.github.com>

* fix: tighten service list locking scope

Agent-Logs-Url: https://github.com/taosdata/TDengine/sessions/5e4db858-6458-40f4-ac28-d1b1b7f97c18

Co-authored-by: hjxilinx <8252296+hjxilinx@users.noreply.github.com>

* fix: restore prophet support and update tests per review feedback

Agent-Logs-Url: https://github.com/taosdata/TDengine/sessions/92298ae1-7da6-4d07-b20e-101c7cd0b26b

Co-authored-by: hjxilinx <8252296+hjxilinx@users.noreply.github.com>

* fix: improve test name and move copy inside lock scope

Agent-Logs-Url: https://github.com/taosdata/TDengine/sessions/92298ae1-7da6-4d07-b20e-101c7cd0b26b

Co-authored-by: hjxilinx <8252296+hjxilinx@users.noreply.github.com>

* Potential fix for pull request finding

Co-au... (continued)

Coverage Stats

238317 of 377957 relevant lines covered (63.05%)

130539817.12 hits per line

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

80.73

/source/dnode/mgmt/node_mgmt/src/dmNodes.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 "dmUtil.h"
#include "monitor.h"
#include "audit.h"

int32_t dmOpenNode(SMgmtWrapper *pWrapper) {
  int32_t code = 0;
  SDnode *pDnode = pWrapper->pDnode;

  if (taosMkDir(pWrapper->path) != 0) {
    code = terrno;
    dError("node:%s, failed to create dir:%s since %s", pWrapper->name, pWrapper->path, tstrerror(code));
    return code;
  }

  SMgmtOutputOpt output = {0};
  SMgmtInputOpt  input = dmBuildMgmtInputOpt(pWrapper);

  dInfo("node:%s, start to open", pWrapper->name);
  tmsgSetDefault(&input.msgCb);
  if ((code = (*pWrapper->func.openFp)(&input, &output)) != 0) {
    dError("node:%s, failed to open since %s", pWrapper->name, tstrerror(code));
    return code;
  }
  dInfo("node:%s, has been opened", pWrapper->name);
  pWrapper->deployed = true;

  if (output.pMgmt != NULL) {
    pWrapper->pMgmt = output.pMgmt;
  }

  dmReportStartup(pWrapper->name, "opened");
  return 0;
}

int32_t dmStartNode(SMgmtWrapper *pWrapper) {
  int32_t code = 0;
  if (pWrapper->func.startFp != NULL) {
    dDebug("node:%s, start to start", pWrapper->name);
    if ((code = (*pWrapper->func.startFp)(pWrapper->pMgmt)) != 0) {
      dError("node:%s, failed to start since %s", pWrapper->name, tstrerror(code));
      return code;
    }
    dDebug("node:%s, has been started", pWrapper->name);
  }

  dmReportStartup(pWrapper->name, "started");
  return 0;
}

void dmStopNode(SMgmtWrapper *pWrapper) {
  if (pWrapper->func.stopFp != NULL && pWrapper->pMgmt != NULL) {
    dDebug("node:%s, start to stop", pWrapper->name);
    (*pWrapper->func.stopFp)(pWrapper->pMgmt);
    dDebug("node:%s, has been stopped", pWrapper->name);
  }
}

void dmCloseNode(SMgmtWrapper *pWrapper) {
  dInfo("node:%s, start to close", pWrapper->name);
  pWrapper->deployed = false;

  while (pWrapper->refCount > 0) {
    taosMsleep(10);
  }

  (void)taosThreadRwlockWrlock(&pWrapper->lock);
  if (pWrapper->pMgmt != NULL) {
    (*pWrapper->func.closeFp)(pWrapper->pMgmt);
    pWrapper->pMgmt = NULL;
  }
  (void)taosThreadRwlockUnlock(&pWrapper->lock);

  dInfo("node:%s, has been closed", pWrapper->name);
}

static int32_t dmOpenNodes(SDnode *pDnode) {
  int32_t code = 0;
  for (EDndNodeType ntype = DNODE; ntype < NODE_END; ++ntype) {
    SMgmtWrapper *pWrapper = &pDnode->wrappers[ntype];
    if (!pWrapper->required) continue;
    if ((code = dmOpenNode(pWrapper)) != 0) {
      dError("node:%s, failed to open since %s", pWrapper->name, tstrerror(code));
      return code;
    }
  }

  auditSetDnodeId(dmGetDnodeId(&pDnode->data));
  monSetDnodeId(dmGetDnodeId(&pDnode->data));

  dmSetStatus(pDnode, DND_STAT_RUNNING);
  return 0;
}

static int32_t dmStartNodes(SDnode *pDnode) {
  int32_t code = 0;
  for (EDndNodeType ntype = DNODE; ntype < NODE_END; ++ntype) {
    SMgmtWrapper *pWrapper = &pDnode->wrappers[ntype];
    if (!pWrapper->required) continue;
    if ((code = dmStartNode(pWrapper)) != 0) {
      dError("node:%s, failed to start since %s", pWrapper->name, tstrerror(code));
      return code;
    }
  }

  dInfo("The daemon initialized successfully");
  dmReportStartup("The daemon", "initialized successfully");
  return 0;
}

static void dmStopNodes(SDnode *pDnode) {
  for (EDndNodeType n = DNODE; n < NODE_END; ++n) {
    SMgmtWrapper *pWrapper = &pDnode->wrappers[n];
    dmStopNode(pWrapper);
  }
}

static void dmCloseNodes(SDnode *pDnode) {
  for (EDndNodeType n = DNODE; n < NODE_END; ++n) {
    SMgmtWrapper *pWrapper = &pDnode->wrappers[n];
    dmCloseNode(pWrapper);
  }
}

int32_t dmRunDnode(SDnode *pDnode) {
  int32_t code = 0;
  int32_t count = 0;

  tsDndStartOsUptime = taosGetOsUptime();

  if ((code = dmOpenNodes(pDnode)) != 0) {
    dError("failed to open nodes since %s", tstrerror(code));
    dmCloseNodes(pDnode);
    return code;
  }

  if ((code = dmStartNodes(pDnode)) != 0) {
    dError("failed to start nodes since %s", tstrerror(code));
    dmSetStatus(pDnode, DND_STAT_STOPPED);
    dmStopNodes(pDnode);
    dmCloseNodes(pDnode);
    return code;
  }
  tsDndStart = taosGetTimestampMs();  // also used to wait for dnode to start
  if (tsDndStart == 0) ++tsDndStart;  // avoid 0

  while (1) {
    if (pDnode->stop) {
      dInfo("The daemon is about to stop");
      dmSetStatus(pDnode, DND_STAT_STOPPED);
      dmStopNodes(pDnode);
      dmCloseNodes(pDnode);
      return 0;
    }

    if (count == 10) {
      if(osUpdate() != 0) {
        dError("failed to update os info");
      }
      count = 0;
    } else {
      count++;
    }

    taosMsleep(100);
  }
}

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 "dmMgmt.h"
18	#include "dmUtil.h"
19	#include "monitor.h"
20	#include "audit.h"
21
22	int32_t dmOpenNode(SMgmtWrapper *pWrapper) {	2,021,399✔
23	int32_t code = 0;	2,021,399✔
24	SDnode *pDnode = pWrapper->pDnode;	2,021,399✔
25
26	if (taosMkDir(pWrapper->path) != 0) {	2,021,399✔
27	code = terrno;	×
28	dError("node:%s, failed to create dir:%s since %s", pWrapper->name, pWrapper->path, tstrerror(code));	×
29	return code;	×
30	}
31
32	SMgmtOutputOpt output = {0};	2,021,399✔
33	SMgmtInputOpt input = dmBuildMgmtInputOpt(pWrapper);	2,021,399✔
34
35	dInfo("node:%s, start to open", pWrapper->name);	2,021,399✔
36	tmsgSetDefault(&input.msgCb);	2,021,399✔
37	if ((code = (*pWrapper->func.openFp)(&input, &output)) != 0) {	2,021,399✔
38	dError("node:%s, failed to open since %s", pWrapper->name, tstrerror(code));	×
39	return code;	×
40	}
41	dInfo("node:%s, has been opened", pWrapper->name);	2,021,399✔
42	pWrapper->deployed = true;	2,021,399✔
43
44	if (output.pMgmt != NULL) {	2,021,399✔
45	pWrapper->pMgmt = output.pMgmt;	2,021,399✔
46	}
47
48	dmReportStartup(pWrapper->name, "opened");	2,021,399✔
49	return 0;	2,021,399✔
50	}
51
52	int32_t dmStartNode(SMgmtWrapper *pWrapper) {	2,021,399✔
53	int32_t code = 0;	2,021,399✔
54	if (pWrapper->func.startFp != NULL) {	2,021,399✔
55	dDebug("node:%s, start to start", pWrapper->name);	1,484,322✔
56	if ((code = (*pWrapper->func.startFp)(pWrapper->pMgmt)) != 0) {	1,484,322✔
57	dError("node:%s, failed to start since %s", pWrapper->name, tstrerror(code));	×
58	return code;	×
59	}
60	dDebug("node:%s, has been started", pWrapper->name);	1,484,322✔
61	}
62
63	dmReportStartup(pWrapper->name, "started");	2,021,399✔
64	return 0;	2,021,399✔
65	}
66
67	void dmStopNode(SMgmtWrapper *pWrapper) {	3,758,879✔
68	if (pWrapper->func.stopFp != NULL && pWrapper->pMgmt != NULL) {	3,758,879✔
69	dDebug("node:%s, start to stop", pWrapper->name);	1,476,752✔
70	(*pWrapper->func.stopFp)(pWrapper->pMgmt);	1,476,752✔
71	dDebug("node:%s, has been stopped", pWrapper->name);	1,476,752✔
72	}
73	}	3,758,879✔
74
75	void dmCloseNode(SMgmtWrapper *pWrapper) {	3,758,879✔
76	dInfo("node:%s, start to close", pWrapper->name);	3,758,879✔
77	pWrapper->deployed = false;	3,758,879✔
78
79	while (pWrapper->refCount > 0) {	3,758,879✔
80	taosMsleep(10);	×
81	}
82
83	(void)taosThreadRwlockWrlock(&pWrapper->lock);	3,758,879✔
84	if (pWrapper->pMgmt != NULL) {	3,758,879✔
85	(*pWrapper->func.closeFp)(pWrapper->pMgmt);	2,021,399✔
86	pWrapper->pMgmt = NULL;	2,021,399✔
87	}
88	(void)taosThreadRwlockUnlock(&pWrapper->lock);	3,758,879✔
89
90	dInfo("node:%s, has been closed", pWrapper->name);	3,758,879✔
91	}	3,758,879✔
92
93	static int32_t dmOpenNodes(SDnode *pDnode) {	536,300✔
94	int32_t code = 0;	536,300✔
95	for (EDndNodeType ntype = DNODE; ntype < NODE_END; ++ntype) {	4,290,400✔
96	SMgmtWrapper *pWrapper = &pDnode->wrappers[ntype];	3,754,100✔
97	if (!pWrapper->required) continue;	3,754,100✔
98	if ((code = dmOpenNode(pWrapper)) != 0) {	2,004,476✔
99	dError("node:%s, failed to open since %s", pWrapper->name, tstrerror(code));	×
100	return code;	×
101	}
102	}
103
104	auditSetDnodeId(dmGetDnodeId(&pDnode->data));	536,300✔
105	monSetDnodeId(dmGetDnodeId(&pDnode->data));	536,300✔
106
107	dmSetStatus(pDnode, DND_STAT_RUNNING);	536,300✔
108	return 0;	536,300✔
109	}
110
111	static int32_t dmStartNodes(SDnode *pDnode) {	536,300✔
112	int32_t code = 0;	536,300✔
113	for (EDndNodeType ntype = DNODE; ntype < NODE_END; ++ntype) {	4,290,400✔
114	SMgmtWrapper *pWrapper = &pDnode->wrappers[ntype];	3,754,100✔
115	if (!pWrapper->required) continue;	3,754,100✔
116	if ((code = dmStartNode(pWrapper)) != 0) {	2,004,476✔
117	dError("node:%s, failed to start since %s", pWrapper->name, tstrerror(code));	×
118	return code;	×
119	}
120	}
121
122	dInfo("The daemon initialized successfully");	536,300✔
123	dmReportStartup("The daemon", "initialized successfully");	536,300✔
124	return 0;	536,300✔
125	}
126
127	static void dmStopNodes(SDnode *pDnode) {	536,300✔
128	for (EDndNodeType n = DNODE; n < NODE_END; ++n) {	4,290,400✔
129	SMgmtWrapper *pWrapper = &pDnode->wrappers[n];	3,754,100✔
130	dmStopNode(pWrapper);	3,754,100✔
131	}
132	}	536,300✔
133
134	static void dmCloseNodes(SDnode *pDnode) {	536,300✔
135	for (EDndNodeType n = DNODE; n < NODE_END; ++n) {	4,290,400✔
136	SMgmtWrapper *pWrapper = &pDnode->wrappers[n];	3,754,100✔
137	dmCloseNode(pWrapper);	3,754,100✔
138	}
139	}	536,300✔
140
141	int32_t dmRunDnode(SDnode *pDnode) {	536,300✔
142	int32_t code = 0;	536,300✔
143	int32_t count = 0;	536,300✔
144
145	tsDndStartOsUptime = taosGetOsUptime();	536,300✔
146
147	if ((code = dmOpenNodes(pDnode)) != 0) {	536,300✔
148	dError("failed to open nodes since %s", tstrerror(code));	×
149	dmCloseNodes(pDnode);	×
150	return code;	×
151	}
152
153	if ((code = dmStartNodes(pDnode)) != 0) {	536,300✔
154	dError("failed to start nodes since %s", tstrerror(code));	×
155	dmSetStatus(pDnode, DND_STAT_STOPPED);	×
156	dmStopNodes(pDnode);	×
157	dmCloseNodes(pDnode);	×
158	return code;	×
159	}
160	tsDndStart = taosGetTimestampMs(); // also used to wait for dnode to start	536,300✔
161	if (tsDndStart == 0) ++tsDndStart; // avoid 0	536,300✔
162
163	while (1) {
164	if (pDnode->stop) {	351,058,916✔
165	dInfo("The daemon is about to stop");	536,300✔
166	dmSetStatus(pDnode, DND_STAT_STOPPED);	536,300✔
167	dmStopNodes(pDnode);	536,300✔
168	dmCloseNodes(pDnode);	536,300✔
169	return 0;	536,300✔
170	}
171
172	if (count == 10) {	350,522,616✔
173	if(osUpdate() != 0) {	31,615,162✔
174	dError("failed to update os info");	×
175	}
176	count = 0;	31,615,162✔
177	} else {
178	count++;	318,907,454✔
179	}
180
181	taosMsleep(100);	350,522,616✔
182	}
183	}

taosdata / TDengine / #5071

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