#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

Run Details

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

6.83

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

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