#4873

Committed 04 Dec 2025 01:55AM UTC coverage: 64.558% (-0.1%) from 64.678%

Build # #4873

Build Type

push

travis-ci

Committed by

guanshengliang

Commit Message

Merge branch '3.0' into cover/3.0

Run Details

719 of 2219 new or added lines in 36 files covered. (32.4%)

6363 existing lines in 135 files now uncovered.

159381 of 246882 relevant lines covered (64.56%)

108937395.15 hits per line

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

43.0

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

/*
 * Copyright (c) 2024 TAOS Data, Inc.
 *
 * 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 <inttypes.h>
#include "mndInstance.h"
#include "mndShow.h"
#include "tcompare.h"
#include "ttime.h"

#define MND_INSTANCE_VER_NUMBER 1

static SSdbRaw *mndInstanceEncode(const SInstanceObj *pInstance);
static SSdbRow *mndInstanceDecode(SSdbRaw *pRaw);
static int32_t  mndInstanceInsert(SSdb *pSdb, SInstanceObj *pInstance);
static int32_t  mndInstanceUpdate(SSdb *pSdb, SInstanceObj *pOld, SInstanceObj *pNew);
static int32_t  mndInstanceDelete(SSdb *pSdb, SInstanceObj *pInstance);
static int32_t  mndRetrieveInstance(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows);
static int32_t  mndProcessInstanceRegister(SRpcMsg *pReq);
static int32_t  mndProcessInstanceList(SRpcMsg *pReq);
static int32_t  mndProcessInstanceTimer(SRpcMsg *pReq);

static inline int32_t mndInstanceDataSize(void) {
  return TSDB_INSTANCE_ID_LEN + TSDB_INSTANCE_TYPE_LEN + TSDB_INSTANCE_DESC_LEN + sizeof(int64_t) * 2 + sizeof(int32_t);
}

int32_t mndInitInstance(SMnode *pMnode) {
  SSdbTable table = {
      .sdbType = SDB_INSTANCE,
      .keyType = SDB_KEY_BINARY,
      .encodeFp = NULL,
      .decodeFp = (SdbDecodeFp)mndInstanceDecode,
      .insertFp = (SdbInsertFp)mndInstanceInsert,
      .updateFp = (SdbUpdateFp)mndInstanceUpdate,
      .deleteFp = (SdbDeleteFp)mndInstanceDelete,
  };

  TAOS_CHECK_RETURN(sdbSetTable(pMnode->pSdb, table));

  mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_INSTANCE, mndRetrieveInstance);
  mndSetMsgHandle(pMnode, TDMT_MND_REGISTER_INSTANCE, mndProcessInstanceRegister);
  mndSetMsgHandle(pMnode, TDMT_MND_LIST_INSTANCES, mndProcessInstanceList);
  mndSetMsgHandle(pMnode, TDMT_MND_INSTANCE_TIMER, mndProcessInstanceTimer);

  return TSDB_CODE_SUCCESS;
}

void mndCleanupInstance(SMnode *pMnode) {
  (void)pMnode;
}

SInstanceObj *mndInstanceAcquire(SMnode *pMnode, const char *id) {
  if (id == NULL || id[0] == 0) {
    terrno = TSDB_CODE_INVALID_PARA;
    return NULL;
  }
  return sdbAcquire(pMnode->pSdb, SDB_INSTANCE, id);
}

void mndInstanceRelease(SMnode *pMnode, SInstanceObj *pInstance) {
  if (pInstance == NULL) return;
  sdbRelease(pMnode->pSdb, pInstance);
}

int32_t mndInstanceUpsert(SMnode *pMnode, const char *id, const char *type, const char *desc, int64_t regTime,
                          int32_t expire) {
  if (id == NULL || id[0] == 0) {
    return TSDB_CODE_INVALID_PARA;
  }

  SInstanceObj instance = {0};
  tstrncpy(instance.id, id, sizeof(instance.id));
  if (type != NULL) {
    tstrncpy(instance.type, type, sizeof(instance.type));
  }
  if (desc != NULL) {
    tstrncpy(instance.desc, desc, sizeof(instance.desc));
  }
  instance.lastRegTime = regTime;
  instance.expire = expire;

  SInstanceObj *pOld = mndInstanceAcquire(pMnode, id);
  if (pOld != NULL) {
    instance.firstRegTime = pOld->firstRegTime;
    if (instance.type[0] == 0) {
      tstrncpy(instance.type, pOld->type, sizeof(instance.type));
    }
    if (instance.desc[0] == 0) {
      tstrncpy(instance.desc, pOld->desc, sizeof(instance.desc));
    }
    mndInstanceRelease(pMnode, pOld);
  } else {
    instance.firstRegTime = regTime;
  }

  SSdbRaw *pRaw = mndInstanceEncode(&instance);
  if (pRaw == NULL) {
    return terrno;
  }

  int32_t code = sdbSetRawStatus(pRaw, SDB_STATUS_READY);
  if (code != TSDB_CODE_SUCCESS) {
    sdbFreeRaw(pRaw);
    return code;
  }

  code = sdbWrite(pMnode->pSdb, pRaw);
  if (code != TSDB_CODE_SUCCESS) {
    mError("instance:%s, failed to upsert into sdb since %s", id, tstrerror(code));
  }
  return code;
}

int32_t mndInstanceRemove(SMnode *pMnode, const char *id) {
  if (id == NULL || id[0] == 0) {
    return TSDB_CODE_INVALID_PARA;
  }

  SInstanceObj instance = {0};
  tstrncpy(instance.id, id, sizeof(instance.id));

  SSdbRaw *pRaw = mndInstanceEncode(&instance);
  if (pRaw == NULL) {
    return terrno;
  }

  int32_t code = sdbSetRawStatus(pRaw, SDB_STATUS_DROPPED);
  if (code != TSDB_CODE_SUCCESS) {
    sdbFreeRaw(pRaw);
    return code;
  }

  code = sdbWrite(pMnode->pSdb, pRaw);
  if (code != TSDB_CODE_SUCCESS) {
    mError("instance:%s, failed to remove from sdb since %s", id, tstrerror(code));
  }
  return code;
}

static SSdbRaw *mndInstanceEncode(const SInstanceObj *pInstance) {
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;
  terrno = TSDB_CODE_SUCCESS;

  SSdbRaw *pRaw = sdbAllocRaw(SDB_INSTANCE, MND_INSTANCE_VER_NUMBER, mndInstanceDataSize());
  if (pRaw == NULL) {
    return NULL;
  }

  int32_t dataPos = 0;
  SDB_SET_BINARY(pRaw, dataPos, pInstance->id, TSDB_INSTANCE_ID_LEN, _OVER);
  SDB_SET_BINARY(pRaw, dataPos, pInstance->type, TSDB_INSTANCE_TYPE_LEN, _OVER);
  SDB_SET_BINARY(pRaw, dataPos, pInstance->desc, TSDB_INSTANCE_DESC_LEN, _OVER);
  SDB_SET_INT64(pRaw, dataPos, pInstance->firstRegTime, _OVER);
  SDB_SET_INT64(pRaw, dataPos, pInstance->lastRegTime, _OVER);
  SDB_SET_INT32(pRaw, dataPos, pInstance->expire, _OVER);
  SDB_SET_DATALEN(pRaw, dataPos, _OVER);

  return pRaw;

_OVER:
  sdbFreeRaw(pRaw);
  terrno = code != TSDB_CODE_SUCCESS ? code : terrno;
  return NULL;
}

static SSdbRow *mndInstanceDecode(SSdbRaw *pRaw) {
  int32_t code = TSDB_CODE_SUCCESS;
  int32_t lino = 0;
  terrno = TSDB_CODE_SUCCESS;

  int8_t sver = 0;
  if (sdbGetRawSoftVer(pRaw, &sver) != 0) {
    return NULL;
  }
  if (sver != MND_INSTANCE_VER_NUMBER) {
    terrno = TSDB_CODE_SDB_INVALID_DATA_VER;
    return NULL;
  }

  SSdbRow *pRow = sdbAllocRow(sizeof(SInstanceObj));
  if (pRow == NULL) {
    terrno = TSDB_CODE_OUT_OF_MEMORY;
    return NULL;
  }

  SInstanceObj *pInstance = sdbGetRowObj(pRow);
  int32_t       dataPos = 0;
  SDB_GET_BINARY(pRaw, dataPos, pInstance->id, TSDB_INSTANCE_ID_LEN, _OVER);
  SDB_GET_BINARY(pRaw, dataPos, pInstance->type, TSDB_INSTANCE_TYPE_LEN, _OVER);
  SDB_GET_BINARY(pRaw, dataPos, pInstance->desc, TSDB_INSTANCE_DESC_LEN, _OVER);
  SDB_GET_INT64(pRaw, dataPos, &pInstance->firstRegTime, _OVER);
  SDB_GET_INT64(pRaw, dataPos, &pInstance->lastRegTime, _OVER);
  SDB_GET_INT32(pRaw, dataPos, &pInstance->expire, _OVER);

  return pRow;

_OVER:
  taosMemoryFreeClear(pRow);
  if (code != TSDB_CODE_SUCCESS) terrno = code;
  return NULL;
}

static int32_t mndInstanceInsert(SSdb *pSdb, SInstanceObj *pInstance) {
  (void)pSdb;
  (void)pInstance;
  return TSDB_CODE_SUCCESS;
}

static int32_t mndInstanceUpdate(SSdb *pSdb, SInstanceObj *pOld, SInstanceObj *pNew) {
  (void)pSdb;
  if (pOld == NULL || pNew == NULL) {
    return TSDB_CODE_INVALID_PARA;
  }
  memcpy(pOld, pNew, sizeof(SInstanceObj));
  return TSDB_CODE_SUCCESS;
}

static int32_t mndInstanceDelete(SSdb *pSdb, SInstanceObj *pInstance) {
  (void)pSdb;
  (void)pInstance;
  return TSDB_CODE_SUCCESS;
}

static int32_t mndProcessInstanceRegister(SRpcMsg *pReq) {
  SMnode              *pMnode = pReq->info.node;
  SInstanceRegisterReq req = {0};
  int32_t              code = tDeserializeSInstanceRegisterReq(pReq->pCont, pReq->contLen, &req);
  if (code != TSDB_CODE_SUCCESS) {
    mError("instance register, failed to decode since %s", tstrerror(code));
    return code;
  }

  if (req.id[0] == 0) {
    mError("instance register, id is empty");
    return TSDB_CODE_INVALID_PARA;
  }

  if (req.expire < 0) {
    code = mndInstanceRemove(pMnode, req.id);
    if (code == TSDB_CODE_SUCCESS) {
      mInfo("instance:%s unregistered", req.id);
    }
  } else {
    int64_t now = taosGetTimestampMs();
    const char *type = req.type[0] ? req.type : NULL;
    const char *desc = req.desc[0] ? req.desc : NULL;
    code = mndInstanceUpsert(pMnode, req.id, type, desc, now, req.expire);
    if (code == TSDB_CODE_SUCCESS) {
      mDebug("instance:%s registered type:%s expire:%d", req.id, req.type, req.expire);
    }
  }

  if (code != TSDB_CODE_SUCCESS) {
    mError("instance:%s, failed to process register request since %s", req.id, tstrerror(code));
  }
  return code;
}

static int32_t mndProcessInstanceList(SRpcMsg *pReq) {
  SMnode          *pMnode = pReq->info.node;
  SSdb            *pSdb = pMnode->pSdb;
  SInstanceListReq req = {0};
  int32_t          code = tDeserializeSInstanceListReq(pReq->pCont, pReq->contLen, &req);
  if (code != TSDB_CODE_SUCCESS) {
    mError("instance list, failed to decode since %s", tstrerror(code));
    return code;
  }

  // Collect instance IDs (keep instances referenced until serialization)
  int32_t        capacity = 32;
  int32_t        count = 0;
  SInstanceObj **instances = taosMemoryCalloc(capacity, sizeof(SInstanceObj *));
  const char   **ids = taosMemoryCalloc(capacity, sizeof(char *));
  void          *iter = NULL;
  SInstanceObj  *pInstance = NULL;
  int64_t        now = taosGetTimestampMs();
  int32_t        filterTypeLen = (req.filter_type[0] != 0) ? (int32_t)strlen(req.filter_type) : 0;

  if (instances == NULL || ids == NULL) {
    taosMemoryFree(instances);
    taosMemoryFree(ids);
    return TSDB_CODE_OUT_OF_MEMORY;
  }

  while (1) {
    iter = sdbFetch(pSdb, SDB_INSTANCE, iter, (void **)&pInstance);
    if (iter == NULL) {
      break;
    }

    // Filter by type if specified
    if (filterTypeLen > 0) {
      int32_t typeLen = (int32_t)strlen(pInstance->type);
      if (typeLen != filterTypeLen || strncasecmp(pInstance->type, req.filter_type, (size_t)filterTypeLen) != 0) {
        sdbRelease(pSdb, pInstance);
        continue;
      }
    }

    // Filter by expiration
    if (pInstance->expire > 0 && pInstance->lastRegTime > 0) {
      int64_t delta = now - pInstance->lastRegTime;
      if (delta > (int64_t)pInstance->expire * 1000) {
        sdbRelease(pSdb, pInstance);
        continue;
      }
    }

    // Add to list (keep reference)
    if (count >= capacity) {
      int32_t newCap = capacity * 2;
      SInstanceObj **newInstances = taosMemoryRealloc(instances, newCap * sizeof(SInstanceObj *));
      if (newInstances == NULL) {
        code = TSDB_CODE_OUT_OF_MEMORY;
        goto _cleanup;
      }
      const char   **newIds = taosMemoryRealloc(ids, newCap * sizeof(char *));
      if (newIds == NULL) {
        // If memory was moved, free newInstances; otherwise, keep instances unchanged
        if (newInstances != instances) {
          // Release instance references in newInstances before freeing the memory
          for (int32_t i = 0; i < count; i++) {
            if (newInstances[i] != NULL) {
              sdbRelease(pSdb, newInstances[i]);
            }
          }
          taosMemoryFree(newInstances);
          instances = NULL;  // Avoid freeing invalid memory in _cleanup
        }
        code = TSDB_CODE_OUT_OF_MEMORY;
        goto _cleanup;
      }
      instances = newInstances;
      ids = newIds;
      capacity = newCap;
    }

    instances[count] = pInstance;
    ids[count] = pInstance->id;
    count++;
  }

  // Serialize response
  SInstanceListRsp rsp = {0};
  rsp.count = count;
  rsp.ids = (char **)ids;
  int32_t contLen = tSerializeSInstanceListRsp(NULL, 0, &rsp);
  if (contLen <= 0) {
    code = terrno != 0 ? terrno : TSDB_CODE_TSC_INTERNAL_ERROR;
    goto _cleanup;
  }

  pReq->info.rsp = rpcMallocCont(contLen);
  if (pReq->info.rsp == NULL) {
    code = TSDB_CODE_OUT_OF_MEMORY;
    goto _cleanup;
  }

  if (tSerializeSInstanceListRsp(pReq->info.rsp, contLen, &rsp) < 0) {
    code = terrno != 0 ? terrno : TSDB_CODE_TSC_INTERNAL_ERROR;
    rpcFreeCont(pReq->info.rsp);
    pReq->info.rsp = NULL;
    goto _cleanup;
  }

  pReq->info.rspLen = contLen;
  code = TSDB_CODE_SUCCESS;

_cleanup:
  // Release all instances
  if (instances != NULL) {
    for (int32_t i = 0; i < count; i++) {
      if (instances[i] != NULL) {
        sdbRelease(pSdb, instances[i]);
      }
    }
    taosMemoryFree(instances);
  }
  if (ids != NULL) {
    taosMemoryFree(ids);
  }
  if (pInstance != NULL) {
    sdbRelease(pSdb, pInstance);
  }

  if (code != TSDB_CODE_SUCCESS) {
    mError("instance list, failed to process request since %s", tstrerror(code));
  } else {
    mDebug("instance list, returned %d instances", count);
  }

  return code;
}

static int32_t mndProcessInstanceTimer(SRpcMsg *pReq) {
  SMnode       *pMnode = pReq->info.node;
  SSdb         *pSdb = pMnode->pSdb;
  void         *iter = NULL;
  SInstanceObj *pInstance = NULL;
  int64_t       now = taosGetTimestampMs();

  while (1) {
    iter = sdbFetch(pSdb, SDB_INSTANCE, iter, (void **)&pInstance);
    if (iter == NULL) {
      break;
    }

    bool expired =
        (pInstance->expire > 0) && (now - pInstance->lastRegTime > ((int64_t)pInstance->expire * 1000));
    if (expired) {
      mInfo("instance:%s expired(last:%" PRId64 "ms, expire:%ds), removing", pInstance->id, pInstance->lastRegTime,
            pInstance->expire);
      (void)mndInstanceRemove(pMnode, pInstance->id);
    }

    sdbRelease(pSdb, pInstance);
  }

  return TSDB_CODE_SUCCESS;
}

static int32_t mndRetrieveInstance(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {
  SMnode *pMnode = pReq->info.node;
  SSdb   *pSdb = pMnode->pSdb;

  int32_t        numOfRows = 0;
  SInstanceObj  *pInstance = NULL;
  int32_t        code = TSDB_CODE_SUCCESS;
  int32_t        lino = 0;

  while (numOfRows < rows) {
    pShow->pIter = sdbFetch(pSdb, SDB_INSTANCE, pShow->pIter, (void **)&pInstance);
    if (pShow->pIter == NULL) break;

    if (pShow->filterTb[0] != 0) {
      if (rawStrPatternMatch(pInstance->id, pShow->filterTb) != TSDB_PATTERN_MATCH) {
        sdbRelease(pSdb, pInstance);
        pInstance = NULL;
        continue;
      }
    }

    int32_t cols = 0;

    SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);

    char idBuf[TSDB_INSTANCE_ID_LEN + VARSTR_HEADER_SIZE] = {0};
    STR_WITH_MAXSIZE_TO_VARSTR(idBuf, pInstance->id, sizeof(idBuf));
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, idBuf, false), pInstance, &lino, _OVER);

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    char typeBuf[TSDB_INSTANCE_TYPE_LEN + VARSTR_HEADER_SIZE] = {0};
    STR_WITH_MAXSIZE_TO_VARSTR(typeBuf, pInstance->type, sizeof(typeBuf));
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, typeBuf, false), pInstance, &lino, _OVER);

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    char descBuf[TSDB_INSTANCE_DESC_LEN + VARSTR_HEADER_SIZE] = {0};
    STR_WITH_MAXSIZE_TO_VARSTR(descBuf, pInstance->desc, sizeof(descBuf));
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, descBuf, false), pInstance, &lino, _OVER);

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pInstance->firstRegTime, false), pInstance,
                        &lino, _OVER);

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pInstance->lastRegTime, false), pInstance,
                        &lino, _OVER);

    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
    RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pInstance->expire, false), pInstance, &lino,
                        _OVER);

    numOfRows++;
    sdbRelease(pSdb, pInstance);
    pInstance = NULL;
  }

_OVER:
  pShow->numOfRows += numOfRows;
  pBlock->info.rows += numOfRows;

  if (code != TSDB_CODE_SUCCESS) {
    mError("failed to retrieve instance rows at line:%d since %s", lino, tstrerror(code));
  }

  return numOfRows;
}


1	/*
2	* Copyright (c) 2024 TAOS Data, Inc.
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 <inttypes.h>
18	#include "mndInstance.h"
19	#include "mndShow.h"
20	#include "tcompare.h"
21	#include "ttime.h"
22
23	#define MND_INSTANCE_VER_NUMBER 1
24
25	static SSdbRaw mndInstanceEncode(const SInstanceObj pInstance);
26	static SSdbRow mndInstanceDecode(SSdbRaw pRaw);
27	static int32_t mndInstanceInsert(SSdb pSdb, SInstanceObj pInstance);
28	static int32_t mndInstanceUpdate(SSdb pSdb, SInstanceObj pOld, SInstanceObj *pNew);
29	static int32_t mndInstanceDelete(SSdb pSdb, SInstanceObj pInstance);
30	static int32_t mndRetrieveInstance(SRpcMsg pReq, SShowObj pShow, SSDataBlock *pBlock, int32_t rows);
31	static int32_t mndProcessInstanceRegister(SRpcMsg *pReq);
32	static int32_t mndProcessInstanceList(SRpcMsg *pReq);
33	static int32_t mndProcessInstanceTimer(SRpcMsg *pReq);
34
35	static inline int32_t mndInstanceDataSize(void) {	32,948✔
36	return TSDB_INSTANCE_ID_LEN + TSDB_INSTANCE_TYPE_LEN + TSDB_INSTANCE_DESC_LEN + sizeof(int64_t) * 2 + sizeof(int32_t);	32,948✔
37	}
38
39	int32_t mndInitInstance(SMnode *pMnode) {	490,193✔
40	SSdbTable table = {	490,193✔
41	.sdbType = SDB_INSTANCE,
42	.keyType = SDB_KEY_BINARY,
43	.encodeFp = NULL,
44	.decodeFp = (SdbDecodeFp)mndInstanceDecode,
45	.insertFp = (SdbInsertFp)mndInstanceInsert,
46	.updateFp = (SdbUpdateFp)mndInstanceUpdate,
47	.deleteFp = (SdbDeleteFp)mndInstanceDelete,
48	};
49
50	TAOS_CHECK_RETURN(sdbSetTable(pMnode->pSdb, table));	490,193✔
51
52	mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_INSTANCE, mndRetrieveInstance);	490,193✔
53	mndSetMsgHandle(pMnode, TDMT_MND_REGISTER_INSTANCE, mndProcessInstanceRegister);	490,193✔
54	mndSetMsgHandle(pMnode, TDMT_MND_LIST_INSTANCES, mndProcessInstanceList);	490,193✔
55	mndSetMsgHandle(pMnode, TDMT_MND_INSTANCE_TIMER, mndProcessInstanceTimer);	490,193✔
56
57	return TSDB_CODE_SUCCESS;	490,193✔
58	}
59
60	void mndCleanupInstance(SMnode *pMnode) {	489,408✔
61	(void)pMnode;
62	}	489,408✔
63
64	SInstanceObj mndInstanceAcquire(SMnode pMnode, const char *id) {	32,947✔
65	if (id == NULL \|\| id[0] == 0) {	32,947✔
NEW 66	terrno = TSDB_CODE_INVALID_PARA;	×
NEW 67	return NULL;	×
68	}
69	return sdbAcquire(pMnode->pSdb, SDB_INSTANCE, id);	32,947✔
70	}
71
72	void mndInstanceRelease(SMnode pMnode, SInstanceObj pInstance) {	30,303✔
73	if (pInstance == NULL) return;	30,303✔
74	sdbRelease(pMnode->pSdb, pInstance);	30,303✔
75	}
76
77	int32_t mndInstanceUpsert(SMnode pMnode, const char id, const char type, const char desc, int64_t regTime,	32,947✔
78	int32_t expire) {
79	if (id == NULL \|\| id[0] == 0) {	32,947✔
NEW 80	return TSDB_CODE_INVALID_PARA;	×
81	}
82
83	SInstanceObj instance = {0};	32,947✔
84	tstrncpy(instance.id, id, sizeof(instance.id));	32,947✔
85	if (type != NULL) {	32,947✔
86	tstrncpy(instance.type, type, sizeof(instance.type));	32,947✔
87	}
88	if (desc != NULL) {	32,947✔
NEW 89	tstrncpy(instance.desc, desc, sizeof(instance.desc));	×
90	}
91	instance.lastRegTime = regTime;	32,947✔
92	instance.expire = expire;	32,947✔
93
94	SInstanceObj *pOld = mndInstanceAcquire(pMnode, id);	32,947✔
95	if (pOld != NULL) {	32,947✔
96	instance.firstRegTime = pOld->firstRegTime;	30,303✔
97	if (instance.type[0] == 0) {	30,303✔
NEW 98	tstrncpy(instance.type, pOld->type, sizeof(instance.type));	×
99	}
100	if (instance.desc[0] == 0) {	30,303✔
101	tstrncpy(instance.desc, pOld->desc, sizeof(instance.desc));	30,303✔
102	}
103	mndInstanceRelease(pMnode, pOld);	30,303✔
104	} else {
105	instance.firstRegTime = regTime;	2,644✔
106	}
107
108	SSdbRaw *pRaw = mndInstanceEncode(&instance);	32,947✔
109	if (pRaw == NULL) {	32,947✔
NEW 110	return terrno;	×
111	}
112
113	int32_t code = sdbSetRawStatus(pRaw, SDB_STATUS_READY);	32,947✔
114	if (code != TSDB_CODE_SUCCESS) {	32,947✔
NEW 115	sdbFreeRaw(pRaw);	×
NEW 116	return code;	×
117	}
118
119	code = sdbWrite(pMnode->pSdb, pRaw);	32,947✔
120	if (code != TSDB_CODE_SUCCESS) {	32,947✔
NEW 121	mError("instance:%s, failed to upsert into sdb since %s", id, tstrerror(code));	×
122	}
123	return code;	32,947✔
124	}
125
126	int32_t mndInstanceRemove(SMnode pMnode, const char id) {	1✔
127	if (id == NULL \|\| id[0] == 0) {	1✔
NEW 128	return TSDB_CODE_INVALID_PARA;	×
129	}
130
131	SInstanceObj instance = {0};	1✔
132	tstrncpy(instance.id, id, sizeof(instance.id));	1✔
133
134	SSdbRaw *pRaw = mndInstanceEncode(&instance);	1✔
135	if (pRaw == NULL) {	1✔
NEW 136	return terrno;	×
137	}
138
139	int32_t code = sdbSetRawStatus(pRaw, SDB_STATUS_DROPPED);	1✔
140	if (code != TSDB_CODE_SUCCESS) {	1✔
NEW 141	sdbFreeRaw(pRaw);	×
NEW 142	return code;	×
143	}
144
145	code = sdbWrite(pMnode->pSdb, pRaw);	1✔
146	if (code != TSDB_CODE_SUCCESS) {	1✔
NEW 147	mError("instance:%s, failed to remove from sdb since %s", id, tstrerror(code));	×
148	}
149	return code;	1✔
150	}
151
152	static SSdbRaw mndInstanceEncode(const SInstanceObj pInstance) {	32,948✔
153	int32_t code = TSDB_CODE_SUCCESS;	32,948✔
154	int32_t lino = 0;	32,948✔
155	terrno = TSDB_CODE_SUCCESS;	32,948✔
156
157	SSdbRaw *pRaw = sdbAllocRaw(SDB_INSTANCE, MND_INSTANCE_VER_NUMBER, mndInstanceDataSize());	32,948✔
158	if (pRaw == NULL) {	32,948✔
NEW 159	return NULL;	×
160	}
161
162	int32_t dataPos = 0;	32,948✔
163	SDB_SET_BINARY(pRaw, dataPos, pInstance->id, TSDB_INSTANCE_ID_LEN, _OVER);	32,948✔
164	SDB_SET_BINARY(pRaw, dataPos, pInstance->type, TSDB_INSTANCE_TYPE_LEN, _OVER);	32,948✔
165	SDB_SET_BINARY(pRaw, dataPos, pInstance->desc, TSDB_INSTANCE_DESC_LEN, _OVER);	32,948✔
166	SDB_SET_INT64(pRaw, dataPos, pInstance->firstRegTime, _OVER);	32,948✔
167	SDB_SET_INT64(pRaw, dataPos, pInstance->lastRegTime, _OVER);	32,948✔
168	SDB_SET_INT32(pRaw, dataPos, pInstance->expire, _OVER);	32,948✔
169	SDB_SET_DATALEN(pRaw, dataPos, _OVER);	32,948✔
170
171	return pRaw;	32,948✔
172
NEW 173	_OVER:	×
NEW 174	sdbFreeRaw(pRaw);	×
NEW 175	terrno = code != TSDB_CODE_SUCCESS ? code : terrno;	×
NEW 176	return NULL;	×
177	}
178
179	static SSdbRow mndInstanceDecode(SSdbRaw pRaw) {	32,948✔
180	int32_t code = TSDB_CODE_SUCCESS;	32,948✔
181	int32_t lino = 0;	32,948✔
182	terrno = TSDB_CODE_SUCCESS;	32,948✔
183
184	int8_t sver = 0;	32,948✔
185	if (sdbGetRawSoftVer(pRaw, &sver) != 0) {	32,948✔
NEW 186	return NULL;	×
187	}
188	if (sver != MND_INSTANCE_VER_NUMBER) {	32,948✔
NEW 189	terrno = TSDB_CODE_SDB_INVALID_DATA_VER;	×
NEW 190	return NULL;	×
191	}
192
193	SSdbRow *pRow = sdbAllocRow(sizeof(SInstanceObj));	32,948✔
194	if (pRow == NULL) {	32,948✔
NEW 195	terrno = TSDB_CODE_OUT_OF_MEMORY;	×
NEW 196	return NULL;	×
197	}
198
199	SInstanceObj *pInstance = sdbGetRowObj(pRow);	32,948✔
200	int32_t dataPos = 0;	32,948✔
201	SDB_GET_BINARY(pRaw, dataPos, pInstance->id, TSDB_INSTANCE_ID_LEN, _OVER);	32,948✔
202	SDB_GET_BINARY(pRaw, dataPos, pInstance->type, TSDB_INSTANCE_TYPE_LEN, _OVER);	32,948✔
203	SDB_GET_BINARY(pRaw, dataPos, pInstance->desc, TSDB_INSTANCE_DESC_LEN, _OVER);	32,948✔
204	SDB_GET_INT64(pRaw, dataPos, &pInstance->firstRegTime, _OVER);	32,948✔
205	SDB_GET_INT64(pRaw, dataPos, &pInstance->lastRegTime, _OVER);	32,948✔
206	SDB_GET_INT32(pRaw, dataPos, &pInstance->expire, _OVER);	32,948✔
207
208	return pRow;	32,948✔
209
NEW 210	_OVER:	×
NEW 211	taosMemoryFreeClear(pRow);	×
NEW 212	if (code != TSDB_CODE_SUCCESS) terrno = code;	×
NEW 213	return NULL;	×
214	}
215
216	static int32_t mndInstanceInsert(SSdb pSdb, SInstanceObj pInstance) {	2,644✔
217	(void)pSdb;
218	(void)pInstance;
219	return TSDB_CODE_SUCCESS;	2,644✔
220	}
221
222	static int32_t mndInstanceUpdate(SSdb pSdb, SInstanceObj pOld, SInstanceObj *pNew) {	30,303✔
223	(void)pSdb;
224	if (pOld == NULL \|\| pNew == NULL) {	30,303✔
NEW 225	return TSDB_CODE_INVALID_PARA;	×
226	}
227	memcpy(pOld, pNew, sizeof(SInstanceObj));	30,303✔
228	return TSDB_CODE_SUCCESS;	30,303✔
229	}
230
231	static int32_t mndInstanceDelete(SSdb pSdb, SInstanceObj pInstance) {	32,948✔
232	(void)pSdb;
233	(void)pInstance;
234	return TSDB_CODE_SUCCESS;	32,948✔
235	}
236
237	static int32_t mndProcessInstanceRegister(SRpcMsg *pReq) {	32,947✔
238	SMnode *pMnode = pReq->info.node;	32,947✔
239	SInstanceRegisterReq req = {0};	32,947✔
240	int32_t code = tDeserializeSInstanceRegisterReq(pReq->pCont, pReq->contLen, &req);	32,947✔
241	if (code != TSDB_CODE_SUCCESS) {	32,947✔
NEW 242	mError("instance register, failed to decode since %s", tstrerror(code));	×
NEW 243	return code;	×
244	}
245
246	if (req.id[0] == 0) {	32,947✔
NEW 247	mError("instance register, id is empty");	×
NEW 248	return TSDB_CODE_INVALID_PARA;	×
249	}
250
251	if (req.expire < 0) {	32,947✔
NEW 252	code = mndInstanceRemove(pMnode, req.id);	×
NEW 253	if (code == TSDB_CODE_SUCCESS) {	×
NEW 254	mInfo("instance:%s unregistered", req.id);	×
255	}
256	} else {
257	int64_t now = taosGetTimestampMs();	32,947✔
258	const char *type = req.type[0] ? req.type : NULL;	32,947✔
259	const char *desc = req.desc[0] ? req.desc : NULL;	32,947✔
260	code = mndInstanceUpsert(pMnode, req.id, type, desc, now, req.expire);	32,947✔
261	if (code == TSDB_CODE_SUCCESS) {	32,947✔
262	mDebug("instance:%s registered type:%s expire:%d", req.id, req.type, req.expire);	32,947✔
263	}
264	}
265
266	if (code != TSDB_CODE_SUCCESS) {	32,947✔
NEW 267	mError("instance:%s, failed to process register request since %s", req.id, tstrerror(code));	×
268	}
269	return code;	32,947✔
270	}
271
NEW 272	static int32_t mndProcessInstanceList(SRpcMsg *pReq) {	×
NEW 273	SMnode *pMnode = pReq->info.node;	×
NEW 274	SSdb *pSdb = pMnode->pSdb;	×
NEW 275	SInstanceListReq req = {0};	×
NEW 276	int32_t code = tDeserializeSInstanceListReq(pReq->pCont, pReq->contLen, &req);	×
NEW 277	if (code != TSDB_CODE_SUCCESS) {	×
NEW 278	mError("instance list, failed to decode since %s", tstrerror(code));	×
NEW 279	return code;	×
280	}
281
282	// Collect instance IDs (keep instances referenced until serialization)
NEW 283	int32_t capacity = 32;	×
NEW 284	int32_t count = 0;	×
NEW 285	SInstanceObj *instances = taosMemoryCalloc(capacity, sizeof(SInstanceObj ));	×
NEW 286	const char *ids = taosMemoryCalloc(capacity, sizeof(char ));	×
NEW 287	void *iter = NULL;	×
NEW 288	SInstanceObj *pInstance = NULL;	×
NEW 289	int64_t now = taosGetTimestampMs();	×
NEW 290	int32_t filterTypeLen = (req.filter_type[0] != 0) ? (int32_t)strlen(req.filter_type) : 0;	×
291
NEW 292	if (instances == NULL \|\| ids == NULL) {	×
NEW 293	taosMemoryFree(instances);	×
NEW 294	taosMemoryFree(ids);	×
NEW 295	return TSDB_CODE_OUT_OF_MEMORY;	×
296	}
297
298	while (1) {
NEW 299	iter = sdbFetch(pSdb, SDB_INSTANCE, iter, (void **)&pInstance);	×
NEW 300	if (iter == NULL) {	×
NEW 301	break;	×
302	}
303
304	// Filter by type if specified
NEW 305	if (filterTypeLen > 0) {	×
NEW 306	int32_t typeLen = (int32_t)strlen(pInstance->type);	×
NEW 307	if (typeLen != filterTypeLen \|\| strncasecmp(pInstance->type, req.filter_type, (size_t)filterTypeLen) != 0) {	×
NEW 308	sdbRelease(pSdb, pInstance);	×
NEW 309	continue;	×
310	}
311	}
312
313	// Filter by expiration
NEW 314	if (pInstance->expire > 0 && pInstance->lastRegTime > 0) {	×
NEW 315	int64_t delta = now - pInstance->lastRegTime;	×
NEW 316	if (delta > (int64_t)pInstance->expire * 1000) {	×
NEW 317	sdbRelease(pSdb, pInstance);	×
NEW 318	continue;	×
319	}
320	}
321
322	// Add to list (keep reference)
NEW 323	if (count >= capacity) {	×
NEW 324	int32_t newCap = capacity * 2;	×
NEW 325	SInstanceObj *newInstances = taosMemoryRealloc(instances, newCap sizeof(SInstanceObj *));	×
NEW 326	if (newInstances == NULL) {	×
NEW 327	code = TSDB_CODE_OUT_OF_MEMORY;	×
NEW 328	goto _cleanup;	×
329	}
NEW 330	const char *newIds = taosMemoryRealloc(ids, newCap sizeof(char *));	×
NEW 331	if (newIds == NULL) {	×
332	// If memory was moved, free newInstances; otherwise, keep instances unchanged
NEW 333	if (newInstances != instances) {	×
334	// Release instance references in newInstances before freeing the memory
NEW 335	for (int32_t i = 0; i < count; i++) {	×
NEW 336	if (newInstances[i] != NULL) {	×
NEW 337	sdbRelease(pSdb, newInstances[i]);	×
338	}
339	}
NEW 340	taosMemoryFree(newInstances);	×
NEW 341	instances = NULL; // Avoid freeing invalid memory in _cleanup	×
342	}
NEW 343	code = TSDB_CODE_OUT_OF_MEMORY;	×
NEW 344	goto _cleanup;	×
345	}
NEW 346	instances = newInstances;	×
NEW 347	ids = newIds;	×
NEW 348	capacity = newCap;	×
349	}
350
NEW 351	instances[count] = pInstance;	×
NEW 352	ids[count] = pInstance->id;	×
NEW 353	count++;	×
354	}
355
356	// Serialize response
NEW 357	SInstanceListRsp rsp = {0};	×
NEW 358	rsp.count = count;	×
NEW 359	rsp.ids = (char **)ids;	×
NEW 360	int32_t contLen = tSerializeSInstanceListRsp(NULL, 0, &rsp);	×
NEW 361	if (contLen <= 0) {	×
NEW 362	code = terrno != 0 ? terrno : TSDB_CODE_TSC_INTERNAL_ERROR;	×
NEW 363	goto _cleanup;	×
364	}
365
NEW 366	pReq->info.rsp = rpcMallocCont(contLen);	×
NEW 367	if (pReq->info.rsp == NULL) {	×
NEW 368	code = TSDB_CODE_OUT_OF_MEMORY;	×
NEW 369	goto _cleanup;	×
370	}
371
NEW 372	if (tSerializeSInstanceListRsp(pReq->info.rsp, contLen, &rsp) < 0) {	×
NEW 373	code = terrno != 0 ? terrno : TSDB_CODE_TSC_INTERNAL_ERROR;	×
NEW 374	rpcFreeCont(pReq->info.rsp);	×
NEW 375	pReq->info.rsp = NULL;	×
NEW 376	goto _cleanup;	×
377	}
378
NEW 379	pReq->info.rspLen = contLen;	×
NEW 380	code = TSDB_CODE_SUCCESS;	×
381
NEW 382	_cleanup:	×
383	// Release all instances
NEW 384	if (instances != NULL) {	×
NEW 385	for (int32_t i = 0; i < count; i++) {	×
NEW 386	if (instances[i] != NULL) {	×
NEW 387	sdbRelease(pSdb, instances[i]);	×
388	}
389	}
NEW 390	taosMemoryFree(instances);	×
391	}
NEW 392	if (ids != NULL) {	×
NEW 393	taosMemoryFree(ids);	×
394	}
NEW 395	if (pInstance != NULL) {	×
NEW 396	sdbRelease(pSdb, pInstance);	×
397	}
398
NEW 399	if (code != TSDB_CODE_SUCCESS) {	×
NEW 400	mError("instance list, failed to process request since %s", tstrerror(code));	×
401	} else {
NEW 402	mDebug("instance list, returned %d instances", count);	×
403	}
404
NEW 405	return code;	×
406	}
407
408	static int32_t mndProcessInstanceTimer(SRpcMsg *pReq) {	4,618,344✔
409	SMnode *pMnode = pReq->info.node;	4,618,344✔
410	SSdb *pSdb = pMnode->pSdb;	4,618,344✔
411	void *iter = NULL;	4,618,344✔
412	SInstanceObj *pInstance = NULL;	4,618,344✔
413	int64_t now = taosGetTimestampMs();	4,618,344✔
414
415	while (1) {	62,045✔
416	iter = sdbFetch(pSdb, SDB_INSTANCE, iter, (void **)&pInstance);	4,680,389✔
417	if (iter == NULL) {	4,680,389✔
418	break;	4,618,344✔
419	}
420
421	bool expired =	62,045✔
422	(pInstance->expire > 0) && (now - pInstance->lastRegTime > ((int64_t)pInstance->expire * 1000));	62,045✔
423	if (expired) {	62,045✔
424	mInfo("instance:%s expired(last:%" PRId64 "ms, expire:%ds), removing", pInstance->id, pInstance->lastRegTime,	1✔
425	pInstance->expire);
426	(void)mndInstanceRemove(pMnode, pInstance->id);	1✔
427	}
428
429	sdbRelease(pSdb, pInstance);	62,045✔
430	}
431
432	return TSDB_CODE_SUCCESS;	4,618,344✔
433	}
434
NEW 435	static int32_t mndRetrieveInstance(SRpcMsg pReq, SShowObj pShow, SSDataBlock *pBlock, int32_t rows) {	×
NEW 436	SMnode *pMnode = pReq->info.node;	×
NEW 437	SSdb *pSdb = pMnode->pSdb;	×
438
NEW 439	int32_t numOfRows = 0;	×
NEW 440	SInstanceObj *pInstance = NULL;	×
NEW 441	int32_t code = TSDB_CODE_SUCCESS;	×
NEW 442	int32_t lino = 0;	×
443
NEW 444	while (numOfRows < rows) {	×
NEW 445	pShow->pIter = sdbFetch(pSdb, SDB_INSTANCE, pShow->pIter, (void **)&pInstance);	×
NEW 446	if (pShow->pIter == NULL) break;	×
447
NEW 448	if (pShow->filterTb[0] != 0) {	×
NEW 449	if (rawStrPatternMatch(pInstance->id, pShow->filterTb) != TSDB_PATTERN_MATCH) {	×
NEW 450	sdbRelease(pSdb, pInstance);	×
NEW 451	pInstance = NULL;	×
NEW 452	continue;	×
453	}
454	}
455
NEW 456	int32_t cols = 0;	×
457
NEW 458	SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);	×
459
NEW 460	char idBuf[TSDB_INSTANCE_ID_LEN + VARSTR_HEADER_SIZE] = {0};	×
NEW 461	STR_WITH_MAXSIZE_TO_VARSTR(idBuf, pInstance->id, sizeof(idBuf));	×
NEW 462	RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, idBuf, false), pInstance, &lino, _OVER);	×
463
NEW 464	pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);	×
NEW 465	char typeBuf[TSDB_INSTANCE_TYPE_LEN + VARSTR_HEADER_SIZE] = {0};	×
NEW 466	STR_WITH_MAXSIZE_TO_VARSTR(typeBuf, pInstance->type, sizeof(typeBuf));	×
NEW 467	RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, typeBuf, false), pInstance, &lino, _OVER);	×
468
NEW 469	pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);	×
NEW 470	char descBuf[TSDB_INSTANCE_DESC_LEN + VARSTR_HEADER_SIZE] = {0};	×
NEW 471	STR_WITH_MAXSIZE_TO_VARSTR(descBuf, pInstance->desc, sizeof(descBuf));	×
NEW 472	RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, descBuf, false), pInstance, &lino, _OVER);	×
473
NEW 474	pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);	×
NEW 475	RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pInstance->firstRegTime, false), pInstance,	×
476	&lino, _OVER);
477
NEW 478	pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);	×
NEW 479	RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pInstance->lastRegTime, false), pInstance,	×
480	&lino, _OVER);
481
NEW 482	pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);	×
NEW 483	RETRIEVE_CHECK_GOTO(colDataSetVal(pColInfo, numOfRows, (const char *)&pInstance->expire, false), pInstance, &lino,	×
484	_OVER);
485
NEW 486	numOfRows++;	×
NEW 487	sdbRelease(pSdb, pInstance);	×
NEW 488	pInstance = NULL;	×
489	}
490
NEW 491	_OVER:	×
NEW 492	pShow->numOfRows += numOfRows;	×
NEW 493	pBlock->info.rows += numOfRows;	×
494
NEW 495	if (code != TSDB_CODE_SUCCESS) {	×
NEW 496	mError("failed to retrieve instance rows at line:%d since %s", lino, tstrerror(code));	×
497	}
498
NEW 499	return numOfRows;	×
500	}
501

taosdata / TDengine / #4873

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