#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

1.79

/source/libs/txnode/src/txnodeMgmt.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/>.
 */

// clang-format off
#include "uv.h"
#include "os.h"
#include "tarray.h"
#include "tglobal.h"
#include "txnode.h"
#include "txnodeInt.h"
#include "osString.h"

// clang-format on

extern char **environ;

#ifdef WINDOWS
#define XNODED_DEFAULT_PATH_1    "C:\\TDengine"
#define XNODED_DEFAULT_PATH_2    "C:\\TDengine"
#define XNODED_DEFAULT_EXEC_NAME "xnoded"
#define XNODED_DEFAULT_EXEC      "\\xnoded.exe"
#else
#define XNODED_DEFAULT_PATH_1    "/usr/bin"
#define XNODED_DEFAULT_PATH_2    "/usr/local/taos/bin"
#define XNODED_DEFAULT_EXEC_NAME "xnoded"
#define XNODED_DEFAULT_EXEC      "/xnoded"
#endif

#define XNODED_XNODED_PID_NAME ".xnoded.pid"

typedef struct {
  bool         isStarted;
  bool         needCleanUp;
  uv_loop_t    loop;
  uv_thread_t  thread;
  uv_barrier_t barrier;
  uv_process_t process;
  int32_t      spawnErr;
  uv_pipe_t    ctrlPipe;
  uv_async_t   stopAsync;
  int32_t      isStopped;
  int32_t      dnodeId;
  int64_t      clusterId;
  char         userPass[XNODE_USER_PASS_LEN];
  char         token[TSDB_TOKEN_LEN + 1];
  SEp          leaderEp;
} SXnodedData;

SXnodedData xnodedGlobal = {0};

static int32_t xnodeMgmtSpawnXnoded(SXnodedData *pData);

static void getXnodedPidPath(char *pipeName, int32_t size) {
  int32_t len = strlen(tsDataDir);

  if (len > 0 && (tsDataDir[len - 1] != '/' && tsDataDir[len - 1] != '\\')) {
  #ifdef _WIN32
    snprintf(pipeName, size, "%s\\%s", tsDataDir, XNODED_XNODED_PID_NAME);
  #else
    snprintf(pipeName, size, "%s/%s", tsDataDir, XNODED_XNODED_PID_NAME);
  #endif
  } else {
    snprintf(pipeName, size, "%s%s", tsDataDir, XNODED_XNODED_PID_NAME);
  }
  xndDebug("xnode get xnoded pid path:%s", pipeName);
}

static void    xnodeMgmtXnodedExit(uv_process_t *process, int64_t exitStatus, int32_t termSignal) {
  TAOS_XNODED_MGMT_CHECK_PTR_RVOID(process);
  SXnodedData *pData = process->data;
  xndInfo("process xnoded exit with status %" PRId64 ", signal %d, isStopped:%d", exitStatus, termSignal,
          pData->isStopped);
  if (pData == NULL) {
    xndError("xnoded process data is NULL");
    return;
  }
  if ((exitStatus == 0 && (termSignal == 0 || termSignal == SIGINT || termSignal == SIGTERM)) ||
      atomic_load_32(&pData->isStopped)) {
    if (uv_async_send(&pData->stopAsync) != 0) {
      xndError("stop xnoded: failed to send stop async");
    }
    char xnodedPipeSocket[PATH_MAX] = {0};
    getXnodedPipeName(xnodedPipeSocket, PATH_MAX);
    if (0 != unlink(xnodedPipeSocket)) {
      xndWarn("txnode failed to unlink, socket: %s, err:%s", xnodedPipeSocket, terrstr());
    }

    char *pidPath = xnodedPipeSocket;
    memset(pidPath, 0, PATH_MAX);
    getXnodedPidPath(pidPath, PATH_MAX);
    (void)taosRemoveFile(pidPath);
    xndInfo("xnoded process exit success");
  } else {
    xndInfo("xnoded process restart, exit status %" PRId64 ", signal %d", exitStatus, termSignal);
    int32_t code = xnodeMgmtSpawnXnoded(pData);
    if (code != 0) {
      xndError("xnoded process restart failed with code:%d", code);
    }
    uv_sleep(500);
  }
}
void killPreXnoded() {
  char buf[PATH_MAX] = {0};
  getXnodedPidPath(buf, sizeof(buf));

  TdFilePtr pFile = NULL;
  pFile = taosOpenFile(buf, TD_FILE_READ);
  if (pFile == NULL) {
    xndWarn("xnode failed to open xnoded pid file:%s, file may not exist", buf);
    return;
  }
  int64_t readSize = taosReadFile(pFile, buf, sizeof(buf));
  if (readSize <= 0) {
    if (readSize < 0) {
      xndError("xnode failed to read len from file:%p since %s", pFile, terrstr());
    }
    (void)taosCloseFile(&pFile);
    return;
  }
  (void)taosCloseFile(&pFile);

  int32_t pid = taosStr2Int32(buf, NULL, 10);
  int result = uv_kill((uv_pid_t)pid, SIGTERM);
  if (result != 0) {
    if (result != UV_ESRCH) {
      xndError("xnode failed to kill process %d: %s", pid, uv_strerror(result));
    }
    return;
  }
}

void saveXnodedPid(int32_t pid) {
  char buf[PATH_MAX] = {0};
  getXnodedPidPath(buf, sizeof(buf));
  TdFilePtr testFilePtr = taosCreateFile(buf, TD_FILE_CREATE | TD_FILE_WRITE | TD_FILE_READ | TD_FILE_TRUNC);
  snprintf(buf, PATH_MAX, "%d", pid);
  (void)taosWriteFile(testFilePtr, buf, strlen(buf));
  (void)taosCloseFile(&testFilePtr);
}

static void locateXnodedExecFile(char *path) {
  if (tsProcPath == NULL) {
    path[0] = '.';
#ifdef WINDOWS
    GetModuleFileName(NULL, path, PATH_MAX);
#elif defined(_TD_DARWIN_64)
    uint32_t pathSize = sizeof(path);
    _NSGetExecutablePath(path, &pathSize);
#endif
  } else {
    TAOS_STRNCPY(path, tsProcPath, PATH_MAX);
  }

  TAOS_DIRNAME(path);
  if (strlen(path) != 0) {
    TAOS_STRCAT(path, XNODED_DEFAULT_EXEC);
    if (taosCheckExistFile(path)) {
      goto _ok;
    }
    xndDebug("can't find xnoded exec file:%s", path);
    path[0] = '\0';
  }

  TAOS_STRCAT(path, XNODED_DEFAULT_PATH_1);
  TAOS_STRCAT(path, XNODED_DEFAULT_EXEC);
  if (taosCheckExistFile(path)) {
    goto _ok;
  }
  xndDebug("can't find xnoded exec file:%s", path);
  path[0] = '\0';

  TAOS_STRCAT(path, XNODED_DEFAULT_PATH_2);
  TAOS_STRCAT(path, XNODED_DEFAULT_EXEC);
  if (taosCheckExistFile(path)) {
    goto _ok;
  }
  xndDebug("can't find xnoded exec file:%s", path);
  path[0] = '\0';

  path[0] = '.';
  path[1] = '\0';
  TAOS_STRCAT(path, XNODED_DEFAULT_EXEC);
  if (taosCheckExistFile(path)) {
    goto _ok;
  }
  xndDebug("can't find xnoded exec file:%s", path);
  path[0] = '\0';

  TAOS_STRNCPY(path, XNODED_DEFAULT_EXEC_NAME, PATH_MAX);
  xndInfo("can't find xnoded exec file, use default: %s", path);
  return;

_ok:
  xndInfo("find xnoded exec file:%s", path);
  return;
}

static int32_t xnodeMgmtSpawnXnoded(SXnodedData *pData) {
  xndDebug("start to init xnoded");
  TAOS_XNODED_MGMT_CHECK_PTR_RCODE(pData);

  int32_t              err = 0;
  uv_process_options_t options = {0};

  char path[PATH_MAX] = {0};
  locateXnodedExecFile(path);

  xndInfo("xnode mgmt spawn xnoded path: %s", path);
  // char *argsXnoded[] = {path, "-c", configDir, "-d", dnodeId, NULL};
  char *argsXnoded[] = {path, NULL};
  options.args = argsXnoded;
  options.file = path;

  options.exit_cb = xnodeMgmtXnodedExit;

  killPreXnoded();

  char xnodedPipeSocket[PATH_MAX] = {0};
  getXnodedPipeName(xnodedPipeSocket, PATH_MAX);
  if (0 != unlink(xnodedPipeSocket)) {
    xndWarn("txnode failed to unlink, ignore if first time, socket: %s, detail:%s", xnodedPipeSocket, terrstr());
  }

  TAOS_UV_LIB_ERROR_RET(uv_pipe_init(&pData->loop, &pData->ctrlPipe, 1));

  uv_stdio_container_t child_stdio[3];
  child_stdio[0].flags = UV_CREATE_PIPE | UV_READABLE_PIPE;
  child_stdio[0].data.stream = (uv_stream_t *)&pData->ctrlPipe;
  child_stdio[1].flags = UV_IGNORE;
  child_stdio[2].flags = UV_INHERIT_FD;
  child_stdio[2].data.fd = 2;
  options.stdio_count = 3;
  options.stdio = child_stdio;

  options.flags = UV_PROCESS_DETACHED;

  char xnodedCfgDir[PATH_MAX + 32] = {0};
  snprintf(xnodedCfgDir, PATH_MAX + 32, "%s=%s", "XNODED_CFG_DIR", configDir);
  char xnodedLogDir[PATH_MAX + 32] = {0};
  snprintf(xnodedLogDir, PATH_MAX + 32, "%s=%s", "XNODED_LOG_DIR", tsLogDir);
  char dnodeIdEnvItem[128] = {0};
  snprintf(dnodeIdEnvItem, 128, "%s=%s:%d", "XNODED_LEADER_EP", pData->leaderEp.fqdn, pData->leaderEp.port);
  char xnodedUserPass[XNODE_USER_PASS_LEN + 32] = {0};
  if (pData->userPass[0] != '\0') {
    snprintf(xnodedUserPass, XNODE_USER_PASS_LEN + 32, "%s=%s", "XNODED_USER_PASS", pData->userPass);
  }
  char xnodedToken[TSDB_TOKEN_LEN + 32] = {0};
  if (pData->token[0] != '\0') {
    snprintf(xnodedToken, TSDB_TOKEN_LEN + 32, "%s=%s", "XNODED_TOKEN", pData->token);
  }
  char xnodeClusterId[64] = {0};
  snprintf(xnodeClusterId, 64, "%s=%" PRIu64, "XNODED_CLUSTER_ID", pData->clusterId);
  char xnodePipeSocket[PATH_MAX + 32] = {0};
  snprintf(xnodePipeSocket, PATH_MAX + 32, "%s=%s", "XNODED_LISTEN", xnodedPipeSocket);

  char xnodedLogLevel[32] = {0};
  if (xndDebugFlag & DEBUG_INFO) {
    snprintf(xnodedLogLevel, 32, "%s=%s", "XNODED_LOG_LEVEL", "info");
  }
  if (xndDebugFlag & DEBUG_DEBUG) {
    snprintf(xnodedLogLevel, 32, "%s=%s", "XNODED_LOG_LEVEL", "debug");
  }
  if (xndDebugFlag & DEBUG_TRACE) {
    snprintf(xnodedLogLevel, 32, "%s=%s", "XNODED_LOG_LEVEL", "trace");
  }

  xndDebug("txnode env: leader ep: %s, pipe socket:%s, log level:%s, cluster_id: %s", dnodeIdEnvItem, xnodePipeSocket,
           xnodedLogLevel, xnodeClusterId);

  char *envXnoded[] = {xnodedCfgDir,    xnodedLogDir,   dnodeIdEnvItem,
                       xnodedUserPass,  xnodedToken,    xnodeClusterId,
                       xnodePipeSocket, xnodedLogLevel, NULL};

  char **envXnodedWithPEnv = NULL;
  if (environ != NULL) {
    int32_t lenEnvXnoded = ARRAY_SIZE(envXnoded);
    int32_t numEnviron = 0;
    while (environ[numEnviron] != NULL) {
      numEnviron++;
    }

    envXnodedWithPEnv = (char **)taosMemoryCalloc(numEnviron + lenEnvXnoded, sizeof(char *));
    if (envXnodedWithPEnv == NULL) {
      err = TSDB_CODE_OUT_OF_MEMORY;
      goto _OVER;
    }

    for (int32_t i = 0; i < numEnviron; i++) {
      int32_t len = strlen(environ[i]) + 1;
      xndDebug("xnoded exec env: %s", environ[i]);
      envXnodedWithPEnv[i] = (char *)taosMemoryCalloc(len, 1);
      if (envXnodedWithPEnv[i] == NULL) {
        err = TSDB_CODE_OUT_OF_MEMORY;
        goto _OVER;
      }

      tstrncpy(envXnodedWithPEnv[i], environ[i], len);
    }

    for (int32_t i = 0, j = 0; i < lenEnvXnoded; i++) {
      if (envXnoded[i] != NULL && envXnoded[i][0] != '\0') {
        int32_t len = strlen(envXnoded[i]) + 1;
        envXnodedWithPEnv[numEnviron + j] = (char *)taosMemoryCalloc(len, 1);
        if (envXnodedWithPEnv[numEnviron + j] == NULL) {
          err = TSDB_CODE_OUT_OF_MEMORY;
          goto _OVER;
        }
        tstrncpy(envXnodedWithPEnv[numEnviron + j], envXnoded[i], len);
        j++;
      }
    }
    envXnodedWithPEnv[numEnviron + lenEnvXnoded - 1] = NULL;

    options.env = envXnodedWithPEnv;
  } else {
    options.env = envXnoded;
  }

  err = uv_spawn(&pData->loop, &pData->process, &options);
  pData->process.data = (void *)pData;
  if (err != 0) {
    xndError("can not spawn xnoded. path: %s, error: %s", path, uv_strerror(err));
  } else {
    xndInfo("xnoded is initialized, xnoded pid: %d", pData->process.pid);
    saveXnodedPid(pData->process.pid);
  }

_OVER:
  if (envXnodedWithPEnv != NULL) {
    int32_t i = 0;
    while (envXnodedWithPEnv[i] != NULL) {
      taosMemoryFree(envXnodedWithPEnv[i]);
      i++;
    }
    taosMemoryFree(envXnodedWithPEnv);
  }

  return err;
}

static void xnodeMgmtXnodedCloseWalkCb(uv_handle_t *handle, void *arg) {
  TAOS_XNODED_MGMT_CHECK_PTR_RVOID(handle);
  if (!uv_is_closing(handle)) {
    xndDebug("xnoded closing handle type:%d, ptr:%p", handle->type, handle);
    uv_close(handle, NULL);
  }
}

static void xnodeMgmtXnodedStopAsyncCb(uv_async_t *async) {
  TAOS_XNODED_MGMT_CHECK_PTR_RVOID(async);
  SXnodedData *pData = async->data;
  uv_stop(&pData->loop);
}

static void xnodeMgmtWatchXnoded(void *args) {
  TAOS_XNODED_MGMT_CHECK_PTR_RVOID(args);
  SXnodedData *pData = args;
  TAOS_UV_CHECK_ERRNO(uv_loop_init(&pData->loop));
  TAOS_UV_CHECK_ERRNO(uv_async_init(&pData->loop, &pData->stopAsync, xnodeMgmtXnodedStopAsyncCb));
  pData->stopAsync.data = pData;
  TAOS_UV_CHECK_ERRNO(xnodeMgmtSpawnXnoded(pData));
  atomic_store_32(&pData->spawnErr, 0);
  (void)uv_barrier_wait(&pData->barrier);
  int32_t num = uv_run(&pData->loop, UV_RUN_DEFAULT);
  xndInfo("xnoded loop exit with %d active handles, line:%d", num, __LINE__);

  uv_walk(&pData->loop, xnodeMgmtXnodedCloseWalkCb, NULL);
  num = uv_run(&pData->loop, UV_RUN_DEFAULT);
  xndInfo("xnoded loop exit with %d active handles, line:%d", num, __LINE__);
  if (uv_loop_close(&pData->loop) != 0) {
    xndError("xnoded loop close failed, lino:%d", __LINE__);
  }
  return;

_exit:
  if (terrno != 0) {
    (void)uv_barrier_wait(&pData->barrier);
    atomic_store_32(&pData->spawnErr, terrno);
    if (uv_loop_close(&pData->loop) != 0) {
      xndError("xnoded loop close failed, lino:%d", __LINE__);
    }

    xndError("xnoded thread exit with code:%d lino:%d", terrno, __LINE__);
    terrno = TSDB_CODE_XNODE_UV_EXEC_FAILURE;
  }
}

/**
 * start xnoded that serves xnode function invocation under dnode startDnodeId
 * @param startDnodeId
 * @return
 */
int32_t xnodeMgmtStartXnoded(SXnode *pXnode) {
  int32_t code = 0, lino = 0;

  SXnodedData *pData = &xnodedGlobal;
  pData->leaderEp = pXnode->ep;
  if (pData->isStarted) {
    xndInfo("dnode start xnoded already called");
    return 0;
  }
  pData->isStarted = true;
  char dnodeId[8] = {0};
  snprintf(dnodeId, sizeof(dnodeId), "%d", pXnode->dnodeId);
  TAOS_CHECK_GOTO(uv_os_setenv("DNODE_ID", dnodeId), &lino, _exit);
  pData->dnodeId = pXnode->dnodeId;
  pData->clusterId = pXnode->clusterId;
  memset(pData->userPass, 0, sizeof(pData->userPass));
  memcpy(pData->userPass, pXnode->userPass, pXnode->upLen);
  memset(pData->token, 0, sizeof(pData->token));
  memcpy(pData->token, pXnode->token, TSDB_TOKEN_LEN);

  TAOS_CHECK_GOTO(uv_barrier_init(&pData->barrier, 2), &lino, _exit);
  TAOS_CHECK_GOTO(uv_thread_create(&pData->thread, xnodeMgmtWatchXnoded, pData), &lino, _exit);
  (void)uv_barrier_wait(&pData->barrier);
  uv_sleep(100);
  int32_t err = atomic_load_32(&pData->spawnErr);
  if (err != 0) {
    uv_barrier_destroy(&pData->barrier);
    if (uv_async_send(&pData->stopAsync) != 0) {
      xndError("start xnoded: failed to send stop async");
    }
    if (uv_thread_join(&pData->thread) != 0) {
      xndError("start xnoded: failed to join xnoded thread");
    }
    pData->needCleanUp = false;
    xndInfo("xnoded is cleaned up after spawn err");
    TAOS_CHECK_GOTO(err, &lino, _exit);
  } else {
    pData->needCleanUp = true;
    atomic_store_32(&pData->isStopped, 0);
  }
_exit:
  if (code != 0) {
    xndError("xnoded start failed with lino:%d, code:%d, error: %s", code, lino, uv_strerror(code));
  }
  return code;
}
/**
 * stop xnoded
 * @return
 */
void xnodeMgmtStopXnoded(void) {
  SXnodedData *pData = &xnodedGlobal;
  xndInfo("stopping xnoded, need cleanup:%d, spawn err:%d, isStopped:%d", pData->needCleanUp, pData->spawnErr, atomic_load_32(&pData->isStopped));
  if (!pData->needCleanUp || atomic_load_32(&pData->isStopped)) {
    return;
  }
  atomic_store_32(&pData->isStopped, 1);
  pData->needCleanUp = false;
  (void)uv_process_kill(&pData->process, SIGTERM);
  uv_sleep(1000);
  uv_barrier_destroy(&pData->barrier);

  xndInfo("xnoded waiting to clean up");
  if (uv_thread_join(&pData->thread) != 0) {
    xndError("stop xnoded: failed to join xnoded thread");
  }
  xndInfo("xnoded is cleaned up");

  pData->isStarted = false;

  return;
}

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	// clang-format off
17	#include "uv.h"
18	#include "os.h"
19	#include "tarray.h"
20	#include "tglobal.h"
21	#include "txnode.h"
22	#include "txnodeInt.h"
23	#include "osString.h"
24
25	// clang-format on
26
27	extern char **environ;
28
29	#ifdef WINDOWS
30	#define XNODED_DEFAULT_PATH_1 "C:\\TDengine"
31	#define XNODED_DEFAULT_PATH_2 "C:\\TDengine"
32	#define XNODED_DEFAULT_EXEC_NAME "xnoded"
33	#define XNODED_DEFAULT_EXEC "\\xnoded.exe"
34	#else
35	#define XNODED_DEFAULT_PATH_1 "/usr/bin"
36	#define XNODED_DEFAULT_PATH_2 "/usr/local/taos/bin"
37	#define XNODED_DEFAULT_EXEC_NAME "xnoded"
38	#define XNODED_DEFAULT_EXEC "/xnoded"
39	#endif
40
41	#define XNODED_XNODED_PID_NAME ".xnoded.pid"
42
43	typedef struct {
44	bool isStarted;
45	bool needCleanUp;
46	uv_loop_t loop;
47	uv_thread_t thread;
48	uv_barrier_t barrier;
49	uv_process_t process;
50	int32_t spawnErr;
51	uv_pipe_t ctrlPipe;
52	uv_async_t stopAsync;
53	int32_t isStopped;
54	int32_t dnodeId;
55	int64_t clusterId;
56	char userPass[XNODE_USER_PASS_LEN];
57	char token[TSDB_TOKEN_LEN + 1];
58	SEp leaderEp;
59	} SXnodedData;
60
61	SXnodedData xnodedGlobal = {0};
62
63	static int32_t xnodeMgmtSpawnXnoded(SXnodedData *pData);
64
65	static void getXnodedPidPath(char *pipeName, int32_t size) {	×
66	int32_t len = strlen(tsDataDir);	×
67
68	if (len > 0 && (tsDataDir[len - 1] != '/' && tsDataDir[len - 1] != '\\')) {	×
69	#ifdef _WIN32
70	snprintf(pipeName, size, "%s\\%s", tsDataDir, XNODED_XNODED_PID_NAME);
71	#else
72	snprintf(pipeName, size, "%s/%s", tsDataDir, XNODED_XNODED_PID_NAME);	×
73	#endif
74	} else {
75	snprintf(pipeName, size, "%s%s", tsDataDir, XNODED_XNODED_PID_NAME);	×
76	}
77	xndDebug("xnode get xnoded pid path:%s", pipeName);	×
78	}	×
79
80	static void xnodeMgmtXnodedExit(uv_process_t *process, int64_t exitStatus, int32_t termSignal) {	×
81	TAOS_XNODED_MGMT_CHECK_PTR_RVOID(process);	×
82	SXnodedData *pData = process->data;	×
83	xndInfo("process xnoded exit with status %" PRId64 ", signal %d, isStopped:%d", exitStatus, termSignal,	×
84	pData->isStopped);
85	if (pData == NULL) {	×
86	xndError("xnoded process data is NULL");	×
87	return;	×
88	}
89	if ((exitStatus == 0 && (termSignal == 0 \|\| termSignal == SIGINT \|\| termSignal == SIGTERM)) \|\|	×
90	atomic_load_32(&pData->isStopped)) {	×
91	if (uv_async_send(&pData->stopAsync) != 0) {	×
92	xndError("stop xnoded: failed to send stop async");	×
93	}
94	char xnodedPipeSocket[PATH_MAX] = {0};	×
95	getXnodedPipeName(xnodedPipeSocket, PATH_MAX);	×
96	if (0 != unlink(xnodedPipeSocket)) {	×
97	xndWarn("txnode failed to unlink, socket: %s, err:%s", xnodedPipeSocket, terrstr());	×
98	}
99
100	char *pidPath = xnodedPipeSocket;	×
101	memset(pidPath, 0, PATH_MAX);	×
102	getXnodedPidPath(pidPath, PATH_MAX);	×
103	(void)taosRemoveFile(pidPath);	×
104	xndInfo("xnoded process exit success");	×
105	} else {
106	xndInfo("xnoded process restart, exit status %" PRId64 ", signal %d", exitStatus, termSignal);	×
107	int32_t code = xnodeMgmtSpawnXnoded(pData);	×
108	if (code != 0) {	×
109	xndError("xnoded process restart failed with code:%d", code);	×
110	}
111	uv_sleep(500);	×
112	}
113	}
114	void killPreXnoded() {	×
115	char buf[PATH_MAX] = {0};	×
116	getXnodedPidPath(buf, sizeof(buf));	×
117
118	TdFilePtr pFile = NULL;	×
119	pFile = taosOpenFile(buf, TD_FILE_READ);	×
120	if (pFile == NULL) {	×
121	xndWarn("xnode failed to open xnoded pid file:%s, file may not exist", buf);	×
122	return;	×
123	}
124	int64_t readSize = taosReadFile(pFile, buf, sizeof(buf));	×
125	if (readSize <= 0) {	×
126	if (readSize < 0) {	×
127	xndError("xnode failed to read len from file:%p since %s", pFile, terrstr());	×
128	}
129	(void)taosCloseFile(&pFile);	×
130	return;	×
131	}
132	(void)taosCloseFile(&pFile);	×
133
134	int32_t pid = taosStr2Int32(buf, NULL, 10);	×
135	int result = uv_kill((uv_pid_t)pid, SIGTERM);	×
136	if (result != 0) {	×
137	if (result != UV_ESRCH) {	×
138	xndError("xnode failed to kill process %d: %s", pid, uv_strerror(result));	×
139	}
140	return;	×
141	}
142	}
143
144	void saveXnodedPid(int32_t pid) {	×
145	char buf[PATH_MAX] = {0};	×
146	getXnodedPidPath(buf, sizeof(buf));	×
147	TdFilePtr testFilePtr = taosCreateFile(buf, TD_FILE_CREATE \| TD_FILE_WRITE \| TD_FILE_READ \| TD_FILE_TRUNC);	×
148	snprintf(buf, PATH_MAX, "%d", pid);	×
149	(void)taosWriteFile(testFilePtr, buf, strlen(buf));	×
150	(void)taosCloseFile(&testFilePtr);	×
151	}	×
152
153	static void locateXnodedExecFile(char *path) {	×
154	if (tsProcPath == NULL) {	×
155	path[0] = '.';	×
156	#ifdef WINDOWS
157	GetModuleFileName(NULL, path, PATH_MAX);
158	#elif defined(_TD_DARWIN_64)
159	uint32_t pathSize = sizeof(path);
160	_NSGetExecutablePath(path, &pathSize);
161	#endif
162	} else {
163	TAOS_STRNCPY(path, tsProcPath, PATH_MAX);	×
164	}
165
166	TAOS_DIRNAME(path);	×
167	if (strlen(path) != 0) {	×
168	TAOS_STRCAT(path, XNODED_DEFAULT_EXEC);	×
169	if (taosCheckExistFile(path)) {	×
170	goto _ok;	×
171	}
172	xndDebug("can't find xnoded exec file:%s", path);	×
173	path[0] = '\0';	×
174	}
175
176	TAOS_STRCAT(path, XNODED_DEFAULT_PATH_1);	×
177	TAOS_STRCAT(path, XNODED_DEFAULT_EXEC);	×
178	if (taosCheckExistFile(path)) {	×
179	goto _ok;	×
180	}
181	xndDebug("can't find xnoded exec file:%s", path);	×
182	path[0] = '\0';	×
183
184	TAOS_STRCAT(path, XNODED_DEFAULT_PATH_2);	×
185	TAOS_STRCAT(path, XNODED_DEFAULT_EXEC);	×
186	if (taosCheckExistFile(path)) {	×
187	goto _ok;	×
188	}
189	xndDebug("can't find xnoded exec file:%s", path);	×
190	path[0] = '\0';	×
191
192	path[0] = '.';	×
193	path[1] = '\0';	×
194	TAOS_STRCAT(path, XNODED_DEFAULT_EXEC);	×
195	if (taosCheckExistFile(path)) {	×
196	goto _ok;	×
197	}
198	xndDebug("can't find xnoded exec file:%s", path);	×
199	path[0] = '\0';	×
200
201	TAOS_STRNCPY(path, XNODED_DEFAULT_EXEC_NAME, PATH_MAX);	×
202	xndInfo("can't find xnoded exec file, use default: %s", path);	×
203	return;	×
204
205	_ok:	×
206	xndInfo("find xnoded exec file:%s", path);	×
207	return;	×
208	}
209
210	static int32_t xnodeMgmtSpawnXnoded(SXnodedData *pData) {	×
211	xndDebug("start to init xnoded");	×
212	TAOS_XNODED_MGMT_CHECK_PTR_RCODE(pData);	×
213
214	int32_t err = 0;	×
215	uv_process_options_t options = {0};	×
216
217	char path[PATH_MAX] = {0};	×
218	locateXnodedExecFile(path);	×
219
220	xndInfo("xnode mgmt spawn xnoded path: %s", path);	×
221	// char *argsXnoded[] = {path, "-c", configDir, "-d", dnodeId, NULL};
222	char *argsXnoded[] = {path, NULL};	×
223	options.args = argsXnoded;	×
224	options.file = path;	×
225
226	options.exit_cb = xnodeMgmtXnodedExit;	×
227
228	killPreXnoded();	×
229
230	char xnodedPipeSocket[PATH_MAX] = {0};	×
231	getXnodedPipeName(xnodedPipeSocket, PATH_MAX);	×
232	if (0 != unlink(xnodedPipeSocket)) {	×
233	xndWarn("txnode failed to unlink, ignore if first time, socket: %s, detail:%s", xnodedPipeSocket, terrstr());	×
234	}
235
236	TAOS_UV_LIB_ERROR_RET(uv_pipe_init(&pData->loop, &pData->ctrlPipe, 1));	×
237
238	uv_stdio_container_t child_stdio[3];	×
239	child_stdio[0].flags = UV_CREATE_PIPE \| UV_READABLE_PIPE;	×
240	child_stdio[0].data.stream = (uv_stream_t *)&pData->ctrlPipe;	×
241	child_stdio[1].flags = UV_IGNORE;	×
242	child_stdio[2].flags = UV_INHERIT_FD;	×
243	child_stdio[2].data.fd = 2;	×
244	options.stdio_count = 3;	×
245	options.stdio = child_stdio;	×
246
247	options.flags = UV_PROCESS_DETACHED;	×
248
249	char xnodedCfgDir[PATH_MAX + 32] = {0};	×
250	snprintf(xnodedCfgDir, PATH_MAX + 32, "%s=%s", "XNODED_CFG_DIR", configDir);	×
251	char xnodedLogDir[PATH_MAX + 32] = {0};	×
252	snprintf(xnodedLogDir, PATH_MAX + 32, "%s=%s", "XNODED_LOG_DIR", tsLogDir);	×
253	char dnodeIdEnvItem[128] = {0};	×
254	snprintf(dnodeIdEnvItem, 128, "%s=%s:%d", "XNODED_LEADER_EP", pData->leaderEp.fqdn, pData->leaderEp.port);	×
255	char xnodedUserPass[XNODE_USER_PASS_LEN + 32] = {0};	×
256	if (pData->userPass[0] != '\0') {	×
257	snprintf(xnodedUserPass, XNODE_USER_PASS_LEN + 32, "%s=%s", "XNODED_USER_PASS", pData->userPass);	×
258	}
259	char xnodedToken[TSDB_TOKEN_LEN + 32] = {0};	×
260	if (pData->token[0] != '\0') {	×
261	snprintf(xnodedToken, TSDB_TOKEN_LEN + 32, "%s=%s", "XNODED_TOKEN", pData->token);	×
262	}
263	char xnodeClusterId[64] = {0};	×
264	snprintf(xnodeClusterId, 64, "%s=%" PRIu64, "XNODED_CLUSTER_ID", pData->clusterId);	×
265	char xnodePipeSocket[PATH_MAX + 32] = {0};	×
266	snprintf(xnodePipeSocket, PATH_MAX + 32, "%s=%s", "XNODED_LISTEN", xnodedPipeSocket);	×
267
268	char xnodedLogLevel[32] = {0};	×
269	if (xndDebugFlag & DEBUG_INFO) {	×
270	snprintf(xnodedLogLevel, 32, "%s=%s", "XNODED_LOG_LEVEL", "info");	×
271	}
272	if (xndDebugFlag & DEBUG_DEBUG) {	×
273	snprintf(xnodedLogLevel, 32, "%s=%s", "XNODED_LOG_LEVEL", "debug");	×
274	}
275	if (xndDebugFlag & DEBUG_TRACE) {	×
276	snprintf(xnodedLogLevel, 32, "%s=%s", "XNODED_LOG_LEVEL", "trace");	×
277	}
278
279	xndDebug("txnode env: leader ep: %s, pipe socket:%s, log level:%s, cluster_id: %s", dnodeIdEnvItem, xnodePipeSocket,	×
280	xnodedLogLevel, xnodeClusterId);
281
282	char *envXnoded[] = {xnodedCfgDir, xnodedLogDir, dnodeIdEnvItem,	×
283	xnodedUserPass, xnodedToken, xnodeClusterId,
284	xnodePipeSocket, xnodedLogLevel, NULL};
285
286	char **envXnodedWithPEnv = NULL;	×
287	if (environ != NULL) {	×
288	int32_t lenEnvXnoded = ARRAY_SIZE(envXnoded);	×
289	int32_t numEnviron = 0;	×
290	while (environ[numEnviron] != NULL) {	×
291	numEnviron++;	×
292	}
293
294	envXnodedWithPEnv = (char *)taosMemoryCalloc(numEnviron + lenEnvXnoded, sizeof(char ));	×
295	if (envXnodedWithPEnv == NULL) {	×
296	err = TSDB_CODE_OUT_OF_MEMORY;	×
297	goto _OVER;	×
298	}
299
300	for (int32_t i = 0; i < numEnviron; i++) {	×
301	int32_t len = strlen(environ[i]) + 1;	×
302	xndDebug("xnoded exec env: %s", environ[i]);	×
303	envXnodedWithPEnv[i] = (char *)taosMemoryCalloc(len, 1);	×
304	if (envXnodedWithPEnv[i] == NULL) {	×
305	err = TSDB_CODE_OUT_OF_MEMORY;	×
306	goto _OVER;	×
307	}
308
309	tstrncpy(envXnodedWithPEnv[i], environ[i], len);	×
310	}
311
312	for (int32_t i = 0, j = 0; i < lenEnvXnoded; i++) {	×
313	if (envXnoded[i] != NULL && envXnoded[i][0] != '\0') {	×
314	int32_t len = strlen(envXnoded[i]) + 1;	×
315	envXnodedWithPEnv[numEnviron + j] = (char *)taosMemoryCalloc(len, 1);	×
316	if (envXnodedWithPEnv[numEnviron + j] == NULL) {	×
317	err = TSDB_CODE_OUT_OF_MEMORY;	×
318	goto _OVER;	×
319	}
320	tstrncpy(envXnodedWithPEnv[numEnviron + j], envXnoded[i], len);	×
321	j++;	×
322	}
323	}
324	envXnodedWithPEnv[numEnviron + lenEnvXnoded - 1] = NULL;	×
325
326	options.env = envXnodedWithPEnv;	×
327	} else {
328	options.env = envXnoded;	×
329	}
330
331	err = uv_spawn(&pData->loop, &pData->process, &options);	×
332	pData->process.data = (void *)pData;	×
333	if (err != 0) {	×
334	xndError("can not spawn xnoded. path: %s, error: %s", path, uv_strerror(err));	×
335	} else {
336	xndInfo("xnoded is initialized, xnoded pid: %d", pData->process.pid);	×
337	saveXnodedPid(pData->process.pid);	×
338	}
339
340	_OVER:	×
341	if (envXnodedWithPEnv != NULL) {	×
342	int32_t i = 0;	×
343	while (envXnodedWithPEnv[i] != NULL) {	×
344	taosMemoryFree(envXnodedWithPEnv[i]);	×
345	i++;	×
346	}
347	taosMemoryFree(envXnodedWithPEnv);	×
348	}
349
350	return err;	×
351	}
352
353	static void xnodeMgmtXnodedCloseWalkCb(uv_handle_t handle, void arg) {	×
354	TAOS_XNODED_MGMT_CHECK_PTR_RVOID(handle);	×
355	if (!uv_is_closing(handle)) {	×
356	xndDebug("xnoded closing handle type:%d, ptr:%p", handle->type, handle);	×
357	uv_close(handle, NULL);	×
358	}
359	}
360
361	static void xnodeMgmtXnodedStopAsyncCb(uv_async_t *async) {	×
362	TAOS_XNODED_MGMT_CHECK_PTR_RVOID(async);	×
363	SXnodedData *pData = async->data;	×
364	uv_stop(&pData->loop);	×
365	}
366
367	static void xnodeMgmtWatchXnoded(void *args) {	×
368	TAOS_XNODED_MGMT_CHECK_PTR_RVOID(args);	×
369	SXnodedData *pData = args;	×
370	TAOS_UV_CHECK_ERRNO(uv_loop_init(&pData->loop));	×
371	TAOS_UV_CHECK_ERRNO(uv_async_init(&pData->loop, &pData->stopAsync, xnodeMgmtXnodedStopAsyncCb));	×
372	pData->stopAsync.data = pData;	×
373	TAOS_UV_CHECK_ERRNO(xnodeMgmtSpawnXnoded(pData));	×
374	atomic_store_32(&pData->spawnErr, 0);	×
375	(void)uv_barrier_wait(&pData->barrier);	×
376	int32_t num = uv_run(&pData->loop, UV_RUN_DEFAULT);	×
377	xndInfo("xnoded loop exit with %d active handles, line:%d", num, __LINE__);	×
378
379	uv_walk(&pData->loop, xnodeMgmtXnodedCloseWalkCb, NULL);	×
380	num = uv_run(&pData->loop, UV_RUN_DEFAULT);	×
381	xndInfo("xnoded loop exit with %d active handles, line:%d", num, __LINE__);	×
382	if (uv_loop_close(&pData->loop) != 0) {	×
383	xndError("xnoded loop close failed, lino:%d", __LINE__);	×
384	}
385	return;	×
386
387	_exit:	×
388	if (terrno != 0) {	×
389	(void)uv_barrier_wait(&pData->barrier);	×
390	atomic_store_32(&pData->spawnErr, terrno);	×
391	if (uv_loop_close(&pData->loop) != 0) {	×
392	xndError("xnoded loop close failed, lino:%d", __LINE__);	×
393	}
394
395	xndError("xnoded thread exit with code:%d lino:%d", terrno, __LINE__);	×
396	terrno = TSDB_CODE_XNODE_UV_EXEC_FAILURE;	×
397	}
398	}
399
400	/**
401	* start xnoded that serves xnode function invocation under dnode startDnodeId
402	* @param startDnodeId
403	* @return
404	*/
405	int32_t xnodeMgmtStartXnoded(SXnode *pXnode) {	×
406	int32_t code = 0, lino = 0;	×
407
408	SXnodedData *pData = &xnodedGlobal;	×
409	pData->leaderEp = pXnode->ep;	×
410	if (pData->isStarted) {	×
411	xndInfo("dnode start xnoded already called");	×
412	return 0;	×
413	}
414	pData->isStarted = true;	×
415	char dnodeId[8] = {0};	×
416	snprintf(dnodeId, sizeof(dnodeId), "%d", pXnode->dnodeId);	×
417	TAOS_CHECK_GOTO(uv_os_setenv("DNODE_ID", dnodeId), &lino, _exit);	×
418	pData->dnodeId = pXnode->dnodeId;	×
419	pData->clusterId = pXnode->clusterId;	×
420	memset(pData->userPass, 0, sizeof(pData->userPass));	×
421	memcpy(pData->userPass, pXnode->userPass, pXnode->upLen);	×
422	memset(pData->token, 0, sizeof(pData->token));	×
423	memcpy(pData->token, pXnode->token, TSDB_TOKEN_LEN);	×
424
425	TAOS_CHECK_GOTO(uv_barrier_init(&pData->barrier, 2), &lino, _exit);	×
426	TAOS_CHECK_GOTO(uv_thread_create(&pData->thread, xnodeMgmtWatchXnoded, pData), &lino, _exit);	×
427	(void)uv_barrier_wait(&pData->barrier);	×
428	uv_sleep(100);	×
429	int32_t err = atomic_load_32(&pData->spawnErr);	×
430	if (err != 0) {	×
431	uv_barrier_destroy(&pData->barrier);	×
432	if (uv_async_send(&pData->stopAsync) != 0) {	×
433	xndError("start xnoded: failed to send stop async");	×
434	}
435	if (uv_thread_join(&pData->thread) != 0) {	×
436	xndError("start xnoded: failed to join xnoded thread");	×
437	}
438	pData->needCleanUp = false;	×
439	xndInfo("xnoded is cleaned up after spawn err");	×
440	TAOS_CHECK_GOTO(err, &lino, _exit);	×
441	} else {
442	pData->needCleanUp = true;	×
443	atomic_store_32(&pData->isStopped, 0);	×
444	}
445	_exit:	×
446	if (code != 0) {	×
447	xndError("xnoded start failed with lino:%d, code:%d, error: %s", code, lino, uv_strerror(code));	×
448	}
449	return code;	×
450	}
451	/**
452	* stop xnoded
453	* @return
454	*/
455	void xnodeMgmtStopXnoded(void) {	549,308✔
456	SXnodedData *pData = &xnodedGlobal;	549,308✔
457	xndInfo("stopping xnoded, need cleanup:%d, spawn err:%d, isStopped:%d", pData->needCleanUp, pData->spawnErr, atomic_load_32(&pData->isStopped));	549,308✔
458	if (!pData->needCleanUp \|\| atomic_load_32(&pData->isStopped)) {	549,308✔
459	return;	549,308✔
460	}
461	atomic_store_32(&pData->isStopped, 1);	×
462	pData->needCleanUp = false;	×
463	(void)uv_process_kill(&pData->process, SIGTERM);	×
464	uv_sleep(1000);	×
465	uv_barrier_destroy(&pData->barrier);	×
466
467	xndInfo("xnoded waiting to clean up");	×
468	if (uv_thread_join(&pData->thread) != 0) {	×
469	xndError("stop xnoded: failed to join xnoded thread");	×
470	}
471	xndInfo("xnoded is cleaned up");	×
472
473	pData->isStarted = false;	×
474
475	return;	×
476	}

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