#4986

Committed 15 Mar 2026 08:32AM UTC coverage: 37.305% (-31.3%) from 68.601%

Build # #4986

Build Type

push

travis-ci

Committed by

tomchon

Commit Message

test: keep docs and unit test

Coverage Stats

125478 of 336361 relevant lines covered (37.3%)

1134847.06 hits per line

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

14.89

/source/dnode/mgmt/mgmt_snode/src/smInt.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 "libs/function/function.h"
#include "libs/function/tudf.h"
#include "smInt.h"
#include "stream.h"
#include "tencrypt.h"

extern SSnodeInfo gSnode;

static int32_t smRequire(const SMgmtInputOpt *pInput, bool *required) {
  char path[TSDB_FILENAME_LEN];
  snprintf(path, TSDB_FILENAME_LEN, "%s%ssnode%d", pInput->path, TD_DIRSEP, pInput->dnodeId);
  SJson    *pJson = NULL;

  int32_t code = dmReadFileJson(path, pInput->name, &pJson, required);
  if (code) {
    return code;
  }

  SDCreateSnodeReq req = {0};
  code = smBuildCreateReqFromJson(pJson, &req);
  if (code) {
    if (pJson != NULL) cJSON_Delete(pJson);
    return code;
  }

  smUpdateSnodeInfo(&req);

  if (pJson != NULL) cJSON_Delete(pJson);
  return code;
}

static void smInitOption(SSnodeMgmt *pMgmt, SSnodeOpt *pOption) { pOption->msgCb = pMgmt->msgCb; }

static void smClose(SSnodeMgmt *pMgmt) {
  if (pMgmt->pSnode != NULL) {
    sndClose(pMgmt->pSnode);
    smStopWorker(pMgmt);
    pMgmt->pSnode = NULL;
  }

  taosMemoryFree(pMgmt);
}
int32_t sndOpenWrapper(const char *path, SSnodeOpt *pOption, SSnode **pNode) {
  *pNode = sndOpen(path, pOption);
  if (*pNode == NULL) {
    return terrno;
  }
  return 0;
}

// Check and migrate checkpoint files to encrypted format if needed
int32_t smCheckAndMigrateCheckpoints(const char *path) {
  int32_t code = TSDB_CODE_SUCCESS;

  // Wait for encryption keys to be loaded
  code = taosWaitCfgKeyLoaded();
  if (TSDB_CODE_SUCCESS != code) {
    dError("failed to wait for encryption keys since %s", tstrerror(code));
    return code;
  }

  // Read encrypted flag from global snode info (set by smBuildCreateReqFromJson)
  taosRLockLatch(&gSnode.snodeLock);
  int32_t encrypted = gSnode.encrypted;
  taosRUnLockLatch(&gSnode.snodeLock);

  // If already encrypted or no metaKey, nothing to do
  if (encrypted || tsMetaKey[0] == '\0') {
    return 0;
  }

  dInfo("snode checkpoints not encrypted but tsMetaKey available, starting migration");

  // List all checkpoint files and re-encrypt them
  char checkpointDir[PATH_MAX] = {0};
  snprintf(checkpointDir, sizeof(checkpointDir), "%s%ssnode%scheckpoint", tsDataDir, TD_DIRSEP, TD_DIRSEP);

  TdDirPtr pDir = taosOpenDir(checkpointDir);
  if (pDir == NULL) {
    // Checkpoint directory doesn't exist yet, just mark as encrypted
    dInfo("checkpoint directory doesn't exist, marking as encrypted");
  } else {
    // Iterate through checkpoint files and re-encrypt each one
    TdDirEntryPtr de = NULL;
    while ((de = taosReadDir(pDir)) != NULL) {
      if (taosDirEntryIsDir(de)) {
        continue;
      }

      char *name = taosGetDirEntryName(de);
      if (strcmp(name, ".") == 0 || strcmp(name, "..") == 0) {
        continue;
      }

      // Only process .ck files
      if (strstr(name, ".ck") == NULL) {
        continue;
      }

      // Extract streamId from filename (format: <streamId>.ck)
      int64_t streamId = 0;
      if (sscanf(name, "%" PRIx64 ".ck", &streamId) != 1) {
        dWarn("skip invalid checkpoint file:%s", name);
        continue;
      }

      dInfo("migrating checkpoint file to encrypted format, streamId:%" PRIx64, streamId);

      // Read checkpoint
      void   *data = NULL;
      int64_t dataLen = 0;
      code = streamReadCheckPoint(streamId, &data, &dataLen);
      if (TSDB_CODE_SUCCESS != code) {
        dError("failed to read checkpoint for migration, streamId:%" PRIx64 " since %s", streamId, tstrerror(code));
        return code;
      }

      if (data == NULL || dataLen <= 0) {
        dInfo("no checkpoint data for streamId:%" PRIx64 ", skipping", streamId);
        taosMemoryFree(data);
        continue;
      }

      // Rewrite checkpoint (will be encrypted with tsMetaKey)
      code = streamWriteCheckPoint(streamId, data, dataLen);
      taosMemoryFree(data);
      if (code != TSDB_CODE_SUCCESS) {
        dError("failed to rewrite checkpoint for streamId:%" PRIx64 " since %s", streamId, tstrerror(code));
        taosMemoryFree(data);
        TAOS_UNUSED(taosCloseDir(&pDir));
        return code;
      }

      dInfo("successfully migrated checkpoint to encrypted format, streamId:%" PRIx64, streamId);
    }

    TAOS_UNUSED(taosCloseDir(&pDir));
  }

  // Update and persist encrypted flag to snode.json
  // First need to read deployed status
  bool   deployed = false;
  SJson *pJson = NULL;
  code = dmReadFileJson(path, "snode", &pJson, &deployed);
  if (TSDB_CODE_SUCCESS != code) {
    dError("failed to read snode.json for persisting encrypted flag since %s", tstrerror(code));
    return code;
  }
  if (pJson != NULL) cJSON_Delete(pJson);

  // Now write with encrypted flag set to true
  code = dmWriteFileWithEncrypted(path, "snode", deployed, true);
  if (TSDB_CODE_SUCCESS != code) {
    dError("failed to persist encrypted flag to snode.json since %s", tstrerror(code));
    return code;
  }

  // Update global encrypted flag
  taosWLockLatch(&gSnode.snodeLock);
  gSnode.encrypted = 1;
  taosWUnLockLatch(&gSnode.snodeLock);

  dInfo("snode checkpoints migrated to encrypted format and persisted encrypted flag successfully");
  return TSDB_CODE_SUCCESS;
}

int32_t smOpen(SMgmtInputOpt *pInput, SMgmtOutputOpt *pOutput) {
  int32_t     code = 0;
  SSnodeMgmt *pMgmt = taosMemoryCalloc(1, sizeof(SSnodeMgmt));
  if (pMgmt == NULL) {
    code = terrno;
    return code;
  }

  pMgmt->pData = pInput->pData;
  pMgmt->path = pInput->path;
  pMgmt->name = pInput->name;
  pMgmt->msgCb = pInput->msgCb;
  pMgmt->msgCb.mgmt = pMgmt;
  pMgmt->msgCb.putToQueueFp = (PutToQueueFp)smPutMsgToQueue;

  SSnodeOpt option = {0};
  smInitOption(pMgmt, &option);

  code = sndOpenWrapper(pMgmt->path, &option, &pMgmt->pSnode);
  if (code != 0) {
    dError("failed to open snode since %s", tstrerror(code));
    smClose(pMgmt);
    return code;
  }

  tmsgReportStartup("snode-impl", "initialized");

  // Check and migrate checkpoint files to encrypted format if needed
  // The encrypted flag is read from snode.json in smRequire() via smBuildCreateReqFromJson()
  code = smCheckAndMigrateCheckpoints(pMgmt->path);
  if (TSDB_CODE_SUCCESS != code) {
    dError("failed to check and migrate snode checkpoints since %s", tstrerror(code));
    // Don't fail the startup, just log the error
    smClose(pMgmt);
    return code;
  }

  if ((code = smStartWorker(pMgmt)) != 0) {
    dError("failed to start snode worker since %s", tstrerror(code));
    smClose(pMgmt);
    return code;
  }
  tmsgReportStartup("snode-worker", "initialized");

  if ((code = udfcOpen()) != 0) {
    dError("failed to open udfc in snode since:%s", tstrerror(code));
    smClose(pMgmt);
    return code;
  }

  pOutput->pMgmt = pMgmt;
  return 0;
}

static int32_t smStartSnodes(SSnodeMgmt *pMgmt) { 
  return sndInit(pMgmt->pSnode); 
}

SMgmtFunc smGetMgmtFunc() {
  SMgmtFunc mgmtFunc = {0};
  mgmtFunc.openFp = smOpen;
  mgmtFunc.startFp = (NodeStartFp)smStartSnodes;
  mgmtFunc.closeFp = (NodeCloseFp)smClose;
  mgmtFunc.createFp = (NodeCreateFp)smProcessCreateReq;
  mgmtFunc.dropFp = (NodeDropFp)smProcessDropReq;
  mgmtFunc.requiredFp = smRequire;
  mgmtFunc.getHandlesFp = smGetMsgHandles;

  return mgmtFunc;
}

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 "libs/function/function.h"
18	#include "libs/function/tudf.h"
19	#include "smInt.h"
20	#include "stream.h"
21	#include "tencrypt.h"
22
23	extern SSnodeInfo gSnode;
24
25	static int32_t smRequire(const SMgmtInputOpt pInput, bool required) {	16✔
26	char path[TSDB_FILENAME_LEN];
27	snprintf(path, TSDB_FILENAME_LEN, "%s%ssnode%d", pInput->path, TD_DIRSEP, pInput->dnodeId);	16✔
28	SJson *pJson = NULL;	16✔
29
30	int32_t code = dmReadFileJson(path, pInput->name, &pJson, required);	16✔
31	if (code) {	16✔
32	return code;	×
33	}
34
35	SDCreateSnodeReq req = {0};	16✔
36	code = smBuildCreateReqFromJson(pJson, &req);	16✔
37	if (code) {	16✔
38	if (pJson != NULL) cJSON_Delete(pJson);	×
39	return code;	×
40	}
41
42	smUpdateSnodeInfo(&req);	16✔
43
44	if (pJson != NULL) cJSON_Delete(pJson);	16✔
45	return code;	16✔
46	}
47
48	static void smInitOption(SSnodeMgmt pMgmt, SSnodeOpt pOption) { pOption->msgCb = pMgmt->msgCb; }	×
49
50	static void smClose(SSnodeMgmt *pMgmt) {	×
51	if (pMgmt->pSnode != NULL) {	×
52	sndClose(pMgmt->pSnode);	×
53	smStopWorker(pMgmt);	×
54	pMgmt->pSnode = NULL;	×
55	}
56
57	taosMemoryFree(pMgmt);	×
58	}	×
59	int32_t sndOpenWrapper(const char path, SSnodeOpt pOption, SSnode **pNode) {	×
60	*pNode = sndOpen(path, pOption);	×
61	if (*pNode == NULL) {	×
62	return terrno;	×
63	}
64	return 0;	×
65	}
66
67	// Check and migrate checkpoint files to encrypted format if needed
68	int32_t smCheckAndMigrateCheckpoints(const char *path) {	×
69	int32_t code = TSDB_CODE_SUCCESS;	×
70
71	// Wait for encryption keys to be loaded
72	code = taosWaitCfgKeyLoaded();	×
73	if (TSDB_CODE_SUCCESS != code) {	×
74	dError("failed to wait for encryption keys since %s", tstrerror(code));	×
75	return code;	×
76	}
77
78	// Read encrypted flag from global snode info (set by smBuildCreateReqFromJson)
79	taosRLockLatch(&gSnode.snodeLock);	×
80	int32_t encrypted = gSnode.encrypted;	×
81	taosRUnLockLatch(&gSnode.snodeLock);	×
82
83	// If already encrypted or no metaKey, nothing to do
84	if (encrypted \|\| tsMetaKey[0] == '\0') {	×
85	return 0;	×
86	}
87
88	dInfo("snode checkpoints not encrypted but tsMetaKey available, starting migration");	×
89
90	// List all checkpoint files and re-encrypt them
91	char checkpointDir[PATH_MAX] = {0};	×
92	snprintf(checkpointDir, sizeof(checkpointDir), "%s%ssnode%scheckpoint", tsDataDir, TD_DIRSEP, TD_DIRSEP);	×
93
94	TdDirPtr pDir = taosOpenDir(checkpointDir);	×
95	if (pDir == NULL) {	×
96	// Checkpoint directory doesn't exist yet, just mark as encrypted
97	dInfo("checkpoint directory doesn't exist, marking as encrypted");	×
98	} else {
99	// Iterate through checkpoint files and re-encrypt each one
100	TdDirEntryPtr de = NULL;	×
101	while ((de = taosReadDir(pDir)) != NULL) {	×
102	if (taosDirEntryIsDir(de)) {	×
103	continue;	×
104	}
105
106	char *name = taosGetDirEntryName(de);	×
107	if (strcmp(name, ".") == 0 \|\| strcmp(name, "..") == 0) {	×
108	continue;	×
109	}
110
111	// Only process .ck files
112	if (strstr(name, ".ck") == NULL) {	×
113	continue;	×
114	}
115
116	// Extract streamId from filename (format: <streamId>.ck)
117	int64_t streamId = 0;	×
118	if (sscanf(name, "%" PRIx64 ".ck", &streamId) != 1) {	×
119	dWarn("skip invalid checkpoint file:%s", name);	×
120	continue;	×
121	}
122
123	dInfo("migrating checkpoint file to encrypted format, streamId:%" PRIx64, streamId);	×
124
125	// Read checkpoint
126	void *data = NULL;	×
127	int64_t dataLen = 0;	×
128	code = streamReadCheckPoint(streamId, &data, &dataLen);	×
129	if (TSDB_CODE_SUCCESS != code) {	×
130	dError("failed to read checkpoint for migration, streamId:%" PRIx64 " since %s", streamId, tstrerror(code));	×
131	return code;	×
132	}
133
134	if (data == NULL \|\| dataLen <= 0) {	×
135	dInfo("no checkpoint data for streamId:%" PRIx64 ", skipping", streamId);	×
136	taosMemoryFree(data);	×
137	continue;	×
138	}
139
140	// Rewrite checkpoint (will be encrypted with tsMetaKey)
141	code = streamWriteCheckPoint(streamId, data, dataLen);	×
142	taosMemoryFree(data);	×
143	if (code != TSDB_CODE_SUCCESS) {	×
144	dError("failed to rewrite checkpoint for streamId:%" PRIx64 " since %s", streamId, tstrerror(code));	×
145	taosMemoryFree(data);	×
146	TAOS_UNUSED(taosCloseDir(&pDir));	×
147	return code;	×
148	}
149
150	dInfo("successfully migrated checkpoint to encrypted format, streamId:%" PRIx64, streamId);	×
151	}
152
153	TAOS_UNUSED(taosCloseDir(&pDir));	×
154	}
155
156	// Update and persist encrypted flag to snode.json
157	// First need to read deployed status
158	bool deployed = false;	×
159	SJson *pJson = NULL;	×
160	code = dmReadFileJson(path, "snode", &pJson, &deployed);	×
161	if (TSDB_CODE_SUCCESS != code) {	×
162	dError("failed to read snode.json for persisting encrypted flag since %s", tstrerror(code));	×
163	return code;	×
164	}
165	if (pJson != NULL) cJSON_Delete(pJson);	×
166
167	// Now write with encrypted flag set to true
168	code = dmWriteFileWithEncrypted(path, "snode", deployed, true);	×
169	if (TSDB_CODE_SUCCESS != code) {	×
170	dError("failed to persist encrypted flag to snode.json since %s", tstrerror(code));	×
171	return code;	×
172	}
173
174	// Update global encrypted flag
175	taosWLockLatch(&gSnode.snodeLock);	×
176	gSnode.encrypted = 1;	×
177	taosWUnLockLatch(&gSnode.snodeLock);	×
178
179	dInfo("snode checkpoints migrated to encrypted format and persisted encrypted flag successfully");	×
180	return TSDB_CODE_SUCCESS;	×
181	}
182
183	int32_t smOpen(SMgmtInputOpt pInput, SMgmtOutputOpt pOutput) {	×
184	int32_t code = 0;	×
185	SSnodeMgmt *pMgmt = taosMemoryCalloc(1, sizeof(SSnodeMgmt));	×
186	if (pMgmt == NULL) {	×
187	code = terrno;	×
188	return code;	×
189	}
190
191	pMgmt->pData = pInput->pData;	×
192	pMgmt->path = pInput->path;	×
193	pMgmt->name = pInput->name;	×
194	pMgmt->msgCb = pInput->msgCb;	×
195	pMgmt->msgCb.mgmt = pMgmt;	×
196	pMgmt->msgCb.putToQueueFp = (PutToQueueFp)smPutMsgToQueue;	×
197
198	SSnodeOpt option = {0};	×
199	smInitOption(pMgmt, &option);	×
200
201	code = sndOpenWrapper(pMgmt->path, &option, &pMgmt->pSnode);	×
202	if (code != 0) {	×
203	dError("failed to open snode since %s", tstrerror(code));	×
204	smClose(pMgmt);	×
205	return code;	×
206	}
207
208	tmsgReportStartup("snode-impl", "initialized");	×
209
210	// Check and migrate checkpoint files to encrypted format if needed
211	// The encrypted flag is read from snode.json in smRequire() via smBuildCreateReqFromJson()
212	code = smCheckAndMigrateCheckpoints(pMgmt->path);	×
213	if (TSDB_CODE_SUCCESS != code) {	×
214	dError("failed to check and migrate snode checkpoints since %s", tstrerror(code));	×
215	// Don't fail the startup, just log the error
216	smClose(pMgmt);	×
217	return code;	×
218	}
219
220	if ((code = smStartWorker(pMgmt)) != 0) {	×
221	dError("failed to start snode worker since %s", tstrerror(code));	×
222	smClose(pMgmt);	×
223	return code;	×
224	}
225	tmsgReportStartup("snode-worker", "initialized");	×
226
227	if ((code = udfcOpen()) != 0) {	×
228	dError("failed to open udfc in snode since:%s", tstrerror(code));	×
229	smClose(pMgmt);	×
230	return code;	×
231	}
232
233	pOutput->pMgmt = pMgmt;	×
234	return 0;	×
235	}
236
237	static int32_t smStartSnodes(SSnodeMgmt *pMgmt) {	×
238	return sndInit(pMgmt->pSnode);	×
239	}
240
241	SMgmtFunc smGetMgmtFunc() {	16✔
242	SMgmtFunc mgmtFunc = {0};	16✔
243	mgmtFunc.openFp = smOpen;	16✔
244	mgmtFunc.startFp = (NodeStartFp)smStartSnodes;	16✔
245	mgmtFunc.closeFp = (NodeCloseFp)smClose;	16✔
246	mgmtFunc.createFp = (NodeCreateFp)smProcessCreateReq;	16✔
247	mgmtFunc.dropFp = (NodeDropFp)smProcessDropReq;	16✔
248	mgmtFunc.requiredFp = smRequire;	16✔
249	mgmtFunc.getHandlesFp = smGetMsgHandles;	16✔
250
251	return mgmtFunc;	16✔
252	}

taosdata / TDengine / #4986

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