提交 4e2a26e8 编写于 作者: S slzhou

Merge branch '3.0' of github.com:taosdata/TDengine into szhou/select-tags-scan

...@@ -1165,6 +1165,9 @@ typedef struct { ...@@ -1165,6 +1165,9 @@ typedef struct {
int32_t vgId; int32_t vgId;
int8_t syncState; int8_t syncState;
int8_t syncRestore; int8_t syncRestore;
int64_t syncTerm;
int64_t roleTimeMs;
int64_t startTimeMs;
int8_t syncCanRead; int8_t syncCanRead;
int64_t cacheUsage; int64_t cacheUsage;
int64_t numOfTables; int64_t numOfTables;
...@@ -1181,9 +1184,9 @@ typedef struct { ...@@ -1181,9 +1184,9 @@ typedef struct {
} SVnodeLoad; } SVnodeLoad;
typedef struct { typedef struct {
int8_t syncState; int8_t syncState;
int8_t syncRestore;
int64_t syncTerm; int64_t syncTerm;
int8_t syncRestore;
int64_t roleTimeMs; int64_t roleTimeMs;
} SMnodeLoad; } SMnodeLoad;
......
...@@ -241,6 +241,7 @@ typedef struct SSyncState { ...@@ -241,6 +241,7 @@ typedef struct SSyncState {
bool canRead; bool canRead;
SyncTerm term; SyncTerm term;
int64_t roleTimeMs; int64_t roleTimeMs;
int64_t startTimeMs;
} SSyncState; } SSyncState;
int32_t syncInit(); int32_t syncInit();
......
...@@ -284,7 +284,6 @@ static const SSysDbTableSchema topicSchema[] = { ...@@ -284,7 +284,6 @@ static const SSysDbTableSchema topicSchema[] = {
{.name = "type", .bytes = 8 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY, .sysInfo = false}, {.name = "type", .bytes = 8 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY, .sysInfo = false},
}; };
static const SSysDbTableSchema subscriptionSchema[] = { static const SSysDbTableSchema subscriptionSchema[] = {
{.name = "topic_name", .bytes = TSDB_TOPIC_FNAME_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY, .sysInfo = false}, {.name = "topic_name", .bytes = TSDB_TOPIC_FNAME_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY, .sysInfo = false},
{.name = "consumer_group", .bytes = TSDB_CGROUP_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY, .sysInfo = false}, {.name = "consumer_group", .bytes = TSDB_CGROUP_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_BINARY, .sysInfo = false},
...@@ -295,12 +294,13 @@ static const SSysDbTableSchema subscriptionSchema[] = { ...@@ -295,12 +294,13 @@ static const SSysDbTableSchema subscriptionSchema[] = {
}; };
static const SSysDbTableSchema vnodesSchema[] = { static const SSysDbTableSchema vnodesSchema[] = {
{.name = "dnode_id", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = true},
{.name = "vgroup_id", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = true}, {.name = "vgroup_id", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = true},
{.name = "replica", .bytes = 1, .type = TSDB_DATA_TYPE_TINYINT, .sysInfo = true}, {.name = "db_name", .bytes = SYSTABLE_SCH_DB_NAME_LEN, .type = TSDB_DATA_TYPE_BINARY, .sysInfo = false},
{.name = "status", .bytes = 9 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = true}, {.name = "status", .bytes = 9 + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = true},
{.name = "db_name", .bytes = SYSTABLE_SCH_DB_NAME_LEN, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = true}, {.name = "role_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP, .sysInfo = true},
{.name = "dnode_id", .bytes = 4, .type = TSDB_DATA_TYPE_INT, .sysInfo = true}, {.name = "start_time", .bytes = 8, .type = TSDB_DATA_TYPE_TIMESTAMP, .sysInfo = true},
{.name = "dnode_ep", .bytes = TSDB_EP_LEN + VARSTR_HEADER_SIZE, .type = TSDB_DATA_TYPE_VARCHAR, .sysInfo = true}, {.name = "restored", .bytes = 1, .type = TSDB_DATA_TYPE_BOOL, .sysInfo = true},
}; };
static const SSysDbTableSchema userUserPrivilegesSchema[] = { static const SSysDbTableSchema userUserPrivilegesSchema[] = {
......
...@@ -1083,8 +1083,8 @@ int32_t tSerializeSStatusReq(void *buf, int32_t bufLen, SStatusReq *pReq) { ...@@ -1083,8 +1083,8 @@ int32_t tSerializeSStatusReq(void *buf, int32_t bufLen, SStatusReq *pReq) {
if (tEncodeI64(&encoder, pload->pointsWritten) < 0) return -1; if (tEncodeI64(&encoder, pload->pointsWritten) < 0) return -1;
if (tEncodeI32(&encoder, pload->numOfCachedTables) < 0) return -1; if (tEncodeI32(&encoder, pload->numOfCachedTables) < 0) return -1;
if (tEncodeI32(&encoder, reserved) < 0) return -1; if (tEncodeI32(&encoder, reserved) < 0) return -1;
if (tEncodeI64(&encoder, reserved) < 0) return -1; if (tEncodeI64(&encoder, pload->roleTimeMs) < 0) return -1;
if (tEncodeI64(&encoder, reserved) < 0) return -1; if (tEncodeI64(&encoder, pload->startTimeMs) < 0) return -1;
} }
// mnode loads // mnode loads
...@@ -1108,6 +1108,16 @@ int32_t tSerializeSStatusReq(void *buf, int32_t bufLen, SStatusReq *pReq) { ...@@ -1108,6 +1108,16 @@ int32_t tSerializeSStatusReq(void *buf, int32_t bufLen, SStatusReq *pReq) {
if (tEncodeI64(&encoder, pReq->mload.syncTerm) < 0) return -1; if (tEncodeI64(&encoder, pReq->mload.syncTerm) < 0) return -1;
if (tEncodeI64(&encoder, pReq->mload.roleTimeMs) < 0) return -1; if (tEncodeI64(&encoder, pReq->mload.roleTimeMs) < 0) return -1;
if (tEncodeI8(&encoder, pReq->clusterCfg.ttlChangeOnWrite) < 0) return -1; if (tEncodeI8(&encoder, pReq->clusterCfg.ttlChangeOnWrite) < 0) return -1;
// vnode extra
for (int32_t i = 0; i < vlen; ++i) {
SVnodeLoad *pload = taosArrayGet(pReq->pVloads, i);
int64_t reserved = 0;
if (tEncodeI64(&encoder, pload->syncTerm) < 0) return -1;
if (tEncodeI64(&encoder, reserved) < 0) return -1;
if (tEncodeI64(&encoder, reserved) < 0) return -1;
if (tEncodeI64(&encoder, reserved) < 0) return -1;
}
tEndEncode(&encoder); tEndEncode(&encoder);
int32_t tlen = encoder.pos; int32_t tlen = encoder.pos;
...@@ -1152,7 +1162,7 @@ int32_t tDeserializeSStatusReq(void *buf, int32_t bufLen, SStatusReq *pReq) { ...@@ -1152,7 +1162,7 @@ int32_t tDeserializeSStatusReq(void *buf, int32_t bufLen, SStatusReq *pReq) {
for (int32_t i = 0; i < vlen; ++i) { for (int32_t i = 0; i < vlen; ++i) {
SVnodeLoad vload = {0}; SVnodeLoad vload = {0};
int64_t reserved64 = 0; vload.syncTerm = -1;
int32_t reserved32 = 0; int32_t reserved32 = 0;
if (tDecodeI32(&decoder, &vload.vgId) < 0) return -1; if (tDecodeI32(&decoder, &vload.vgId) < 0) return -1;
if (tDecodeI8(&decoder, &vload.syncState) < 0) return -1; if (tDecodeI8(&decoder, &vload.syncState) < 0) return -1;
...@@ -1166,14 +1176,15 @@ int32_t tDeserializeSStatusReq(void *buf, int32_t bufLen, SStatusReq *pReq) { ...@@ -1166,14 +1176,15 @@ int32_t tDeserializeSStatusReq(void *buf, int32_t bufLen, SStatusReq *pReq) {
if (tDecodeI64(&decoder, &vload.pointsWritten) < 0) return -1; if (tDecodeI64(&decoder, &vload.pointsWritten) < 0) return -1;
if (tDecodeI32(&decoder, &vload.numOfCachedTables) < 0) return -1; if (tDecodeI32(&decoder, &vload.numOfCachedTables) < 0) return -1;
if (tDecodeI32(&decoder, (int32_t *)&reserved32) < 0) return -1; if (tDecodeI32(&decoder, (int32_t *)&reserved32) < 0) return -1;
if (tDecodeI64(&decoder, &reserved64) < 0) return -1; if (tDecodeI64(&decoder, &vload.roleTimeMs) < 0) return -1;
if (tDecodeI64(&decoder, &reserved64) < 0) return -1; if (tDecodeI64(&decoder, &vload.startTimeMs) < 0) return -1;
if (taosArrayPush(pReq->pVloads, &vload) == NULL) { if (taosArrayPush(pReq->pVloads, &vload) == NULL) {
terrno = TSDB_CODE_OUT_OF_MEMORY; terrno = TSDB_CODE_OUT_OF_MEMORY;
return -1; return -1;
} }
} }
// mnode loads
if (tDecodeI8(&decoder, &pReq->mload.syncState) < 0) return -1; if (tDecodeI8(&decoder, &pReq->mload.syncState) < 0) return -1;
if (tDecodeI8(&decoder, &pReq->mload.syncRestore) < 0) return -1; if (tDecodeI8(&decoder, &pReq->mload.syncRestore) < 0) return -1;
...@@ -1204,6 +1215,17 @@ int32_t tDeserializeSStatusReq(void *buf, int32_t bufLen, SStatusReq *pReq) { ...@@ -1204,6 +1215,17 @@ int32_t tDeserializeSStatusReq(void *buf, int32_t bufLen, SStatusReq *pReq) {
if (tDecodeI8(&decoder, &pReq->clusterCfg.ttlChangeOnWrite) < 0) return -1; if (tDecodeI8(&decoder, &pReq->clusterCfg.ttlChangeOnWrite) < 0) return -1;
} }
// vnode extra
if (!tDecodeIsEnd(&decoder)) {
for (int32_t i = 0; i < vlen; ++i) {
SVnodeLoad *pLoad = taosArrayGet(pReq->pVloads, i);
int64_t reserved = 0;
if (tDecodeI64(&decoder, &pLoad->syncTerm) < 0) return -1;
if (tDecodeI64(&decoder, &reserved) < 0) return -1;
if (tDecodeI64(&decoder, &reserved) < 0) return -1;
if (tDecodeI64(&decoder, &reserved) < 0) return -1;
}
}
tEndDecode(&decoder); tEndDecode(&decoder);
tDecoderClear(&decoder); tDecoderClear(&decoder);
return 0; return 0;
......
...@@ -347,8 +347,11 @@ typedef struct { ...@@ -347,8 +347,11 @@ typedef struct {
typedef struct { typedef struct {
int32_t dnodeId; int32_t dnodeId;
ESyncState syncState; ESyncState syncState;
int64_t syncTerm;
bool syncRestore; bool syncRestore;
bool syncCanRead; bool syncCanRead;
int64_t roleTimeMs;
int64_t startTimeMs;
ESyncRole nodeRole; ESyncRole nodeRole;
} SVnodeGid; } SVnodeGid;
......
...@@ -424,6 +424,47 @@ static int32_t mndCheckClusterCfgPara(SMnode *pMnode, SDnodeObj *pDnode, const S ...@@ -424,6 +424,47 @@ static int32_t mndCheckClusterCfgPara(SMnode *pMnode, SDnodeObj *pDnode, const S
return 0; return 0;
} }
static bool mndUpdateVnodeState(int32_t vgId, SVnodeGid *pGid, SVnodeLoad *pVload) {
bool stateChanged = false;
bool roleChanged = pGid->syncState != pVload->syncState ||
(pVload->syncTerm != -1 && pGid->syncTerm != pVload->syncTerm) ||
pGid->roleTimeMs != pVload->roleTimeMs;
if (roleChanged || pGid->syncRestore != pVload->syncRestore || pGid->syncCanRead != pVload->syncCanRead ||
pGid->startTimeMs != pVload->startTimeMs) {
mInfo(
"vgId:%d, state changed by status msg, old state:%s restored:%d canRead:%d new state:%s restored:%d "
"canRead:%d, dnode:%d",
vgId, syncStr(pGid->syncState), pGid->syncRestore, pGid->syncCanRead, syncStr(pVload->syncState),
pVload->syncRestore, pVload->syncCanRead, pGid->dnodeId);
pGid->syncState = pVload->syncState;
pGid->syncTerm = pVload->syncTerm;
pGid->syncRestore = pVload->syncRestore;
pGid->syncCanRead = pVload->syncCanRead;
pGid->startTimeMs = pVload->startTimeMs;
pGid->roleTimeMs = pVload->roleTimeMs;
stateChanged = true;
}
return stateChanged;
}
static bool mndUpdateMnodeState(SMnodeObj *pObj, SMnodeLoad *pMload) {
bool stateChanged = false;
bool roleChanged = pObj->syncState != pMload->syncState ||
(pMload->syncTerm != -1 && pObj->syncTerm != pMload->syncTerm) ||
pObj->roleTimeMs != pMload->roleTimeMs;
if (roleChanged || pObj->syncRestore != pMload->syncRestore) {
mInfo("dnode:%d, mnode syncState from %s to %s, restoreState from %d to %d, syncTerm from %" PRId64 " to %" PRId64,
pObj->id, syncStr(pObj->syncState), syncStr(pMload->syncState), pObj->syncRestore, pMload->syncRestore,
pObj->syncTerm, pMload->syncTerm);
pObj->syncState = pMload->syncState;
pObj->syncTerm = pMload->syncTerm;
pObj->syncRestore = pMload->syncRestore;
pObj->roleTimeMs = pMload->roleTimeMs;
stateChanged = true;
}
return stateChanged;
}
static int32_t mndProcessStatusReq(SRpcMsg *pReq) { static int32_t mndProcessStatusReq(SRpcMsg *pReq) {
SMnode *pMnode = pReq->info.node; SMnode *pMnode = pReq->info.node;
SStatusReq statusReq = {0}; SStatusReq statusReq = {0};
...@@ -496,26 +537,21 @@ static int32_t mndProcessStatusReq(SRpcMsg *pReq) { ...@@ -496,26 +537,21 @@ static int32_t mndProcessStatusReq(SRpcMsg *pReq) {
pVgroup->compStorage = pVload->compStorage; pVgroup->compStorage = pVload->compStorage;
pVgroup->pointsWritten = pVload->pointsWritten; pVgroup->pointsWritten = pVload->pointsWritten;
} }
bool roleChanged = false; bool stateChanged = false;
for (int32_t vg = 0; vg < pVgroup->replica; ++vg) { for (int32_t vg = 0; vg < pVgroup->replica; ++vg) {
SVnodeGid *pGid = &pVgroup->vnodeGid[vg]; SVnodeGid *pGid = &pVgroup->vnodeGid[vg];
if (pGid->dnodeId == statusReq.dnodeId) { if (pGid->dnodeId == statusReq.dnodeId) {
if (pGid->syncState != pVload->syncState || pGid->syncRestore != pVload->syncRestore || if (pVload->startTimeMs == 0) {
pGid->syncCanRead != pVload->syncCanRead) { pVload->startTimeMs = statusReq.rebootTime;
mInfo( }
"vgId:%d, state changed by status msg, old state:%s restored:%d canRead:%d new state:%s restored:%d " if (pVload->roleTimeMs == 0) {
"canRead:%d, dnode:%d", pVload->roleTimeMs = statusReq.rebootTime;
pVgroup->vgId, syncStr(pGid->syncState), pGid->syncRestore, pGid->syncCanRead,
syncStr(pVload->syncState), pVload->syncRestore, pVload->syncCanRead, pDnode->id);
pGid->syncState = pVload->syncState;
pGid->syncRestore = pVload->syncRestore;
pGid->syncCanRead = pVload->syncCanRead;
roleChanged = true;
} }
stateChanged = mndUpdateVnodeState(pVgroup->vgId, pGid, pVload);
break; break;
} }
} }
if (roleChanged) { if (stateChanged) {
SDbObj *pDb = mndAcquireDb(pMnode, pVgroup->dbName); SDbObj *pDb = mndAcquireDb(pMnode, pVgroup->dbName);
if (pDb != NULL && pDb->stateTs != curMs) { if (pDb != NULL && pDb->stateTs != curMs) {
mInfo("db:%s, stateTs changed by status msg, old stateTs:%" PRId64 " new stateTs:%" PRId64, pDb->name, mInfo("db:%s, stateTs changed by status msg, old stateTs:%" PRId64 " new stateTs:%" PRId64, pDb->name,
...@@ -531,23 +567,10 @@ static int32_t mndProcessStatusReq(SRpcMsg *pReq) { ...@@ -531,23 +567,10 @@ static int32_t mndProcessStatusReq(SRpcMsg *pReq) {
SMnodeObj *pObj = mndAcquireMnode(pMnode, pDnode->id); SMnodeObj *pObj = mndAcquireMnode(pMnode, pDnode->id);
if (pObj != NULL) { if (pObj != NULL) {
bool roleChanged = pObj->syncState != statusReq.mload.syncState || if (statusReq.mload.roleTimeMs == 0) {
(statusReq.mload.syncTerm != -1 && pObj->syncTerm != statusReq.mload.syncTerm); statusReq.mload.roleTimeMs = statusReq.rebootTime;
bool restoreChanged = pObj->syncRestore != statusReq.mload.syncRestore;
if (roleChanged || restoreChanged) {
mInfo("dnode:%d, mnode syncState from %s to %s, restoreState from %d to %d, syncTerm from %" PRId64
" to %" PRId64,
pObj->id, syncStr(pObj->syncState), syncStr(statusReq.mload.syncState), pObj->syncRestore,
statusReq.mload.syncRestore, pObj->syncTerm, statusReq.mload.syncTerm);
pObj->syncState = statusReq.mload.syncState;
pObj->syncRestore = statusReq.mload.syncRestore;
pObj->syncTerm = statusReq.mload.syncTerm;
} }
mndUpdateMnodeState(pObj, &statusReq.mload);
if (roleChanged) {
pObj->roleTimeMs = (statusReq.mload.roleTimeMs != 0) ? statusReq.mload.roleTimeMs : taosGetTimestampMs();
}
mndReleaseMnode(pMnode, pObj); mndReleaseMnode(pMnode, pObj);
} }
......
...@@ -185,7 +185,7 @@ static void mndSetVgroupOffline(SMnode *pMnode, int32_t dnodeId, int64_t curMs) ...@@ -185,7 +185,7 @@ static void mndSetVgroupOffline(SMnode *pMnode, int32_t dnodeId, int64_t curMs)
pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup); pIter = sdbFetch(pSdb, SDB_VGROUP, pIter, (void **)&pVgroup);
if (pIter == NULL) break; if (pIter == NULL) break;
bool roleChanged = false; bool stateChanged = false;
for (int32_t vg = 0; vg < pVgroup->replica; ++vg) { for (int32_t vg = 0; vg < pVgroup->replica; ++vg) {
SVnodeGid *pGid = &pVgroup->vnodeGid[vg]; SVnodeGid *pGid = &pVgroup->vnodeGid[vg];
if (pGid->dnodeId == dnodeId) { if (pGid->dnodeId == dnodeId) {
...@@ -197,13 +197,14 @@ static void mndSetVgroupOffline(SMnode *pMnode, int32_t dnodeId, int64_t curMs) ...@@ -197,13 +197,14 @@ static void mndSetVgroupOffline(SMnode *pMnode, int32_t dnodeId, int64_t curMs)
pGid->syncState = TAOS_SYNC_STATE_OFFLINE; pGid->syncState = TAOS_SYNC_STATE_OFFLINE;
pGid->syncRestore = 0; pGid->syncRestore = 0;
pGid->syncCanRead = 0; pGid->syncCanRead = 0;
roleChanged = true; pGid->startTimeMs = 0;
stateChanged = true;
} }
break; break;
} }
} }
if (roleChanged) { if (stateChanged) {
SDbObj *pDb = mndAcquireDb(pMnode, pVgroup->dbName); SDbObj *pDb = mndAcquireDb(pMnode, pVgroup->dbName);
if (pDb != NULL && pDb->stateTs != curMs) { if (pDb != NULL && pDb->stateTs != curMs) {
mInfo("db:%s, stateTs changed by offline check, old newTs:%" PRId64 " newTs:%" PRId64, pDb->name, pDb->stateTs, mInfo("db:%s, stateTs changed by offline check, old newTs:%" PRId64 " newTs:%" PRId64, pDb->name, pDb->stateTs,
......
...@@ -807,7 +807,6 @@ static int32_t mndRetrieveMnodes(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pB ...@@ -807,7 +807,6 @@ static int32_t mndRetrieveMnodes(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pB
ESdbStatus objStatus = 0; ESdbStatus objStatus = 0;
char *pWrite; char *pWrite;
int64_t curMs = taosGetTimestampMs(); int64_t curMs = taosGetTimestampMs();
int64_t dummyTimeMs = 0;
pSelfObj = sdbAcquire(pSdb, SDB_MNODE, &pMnode->selfDnodeId); pSelfObj = sdbAcquire(pSdb, SDB_MNODE, &pMnode->selfDnodeId);
if (pSelfObj == NULL) { if (pSelfObj == NULL) {
...@@ -858,16 +857,9 @@ static int32_t mndRetrieveMnodes(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pB ...@@ -858,16 +857,9 @@ static int32_t mndRetrieveMnodes(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pB
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataSetVal(pColInfo, numOfRows, (const char *)&pObj->createdTime, false); colDataSetVal(pColInfo, numOfRows, (const char *)&pObj->createdTime, false);
int64_t roleTimeMs = (isDnodeOnline) ? pObj->roleTimeMs : 0;
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
if (pObj->syncTerm != pSelfObj->syncTerm || !isDnodeOnline) { colDataSetVal(pColInfo, numOfRows, (const char *)&roleTimeMs, false);
// state of old term / no status report => use dummyTimeMs
if (pObj->syncTerm > pSelfObj->syncTerm) {
mError("mnode:%d has a newer term:%" PRId64 " than me:%" PRId64, pObj->id, pObj->syncTerm, pSelfObj->syncTerm);
}
colDataSetVal(pColInfo, numOfRows, (const char *)&dummyTimeMs, false);
} else {
colDataSetVal(pColInfo, numOfRows, (const char *)&pObj->roleTimeMs, false);
}
numOfRows++; numOfRows++;
sdbRelease(pSdb, pObj); sdbRelease(pSdb, pObj);
......
...@@ -961,27 +961,24 @@ static int32_t mndRetrieveVnodes(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pB ...@@ -961,27 +961,24 @@ static int32_t mndRetrieveVnodes(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pB
int32_t numOfRows = 0; int32_t numOfRows = 0;
SVgObj *pVgroup = NULL; SVgObj *pVgroup = NULL;
int32_t cols = 0; int32_t cols = 0;
int64_t curMs = taosGetTimestampMs();
while (numOfRows < rows) { while (numOfRows < rows) {
pShow->pIter = sdbFetch(pSdb, SDB_VGROUP, pShow->pIter, (void **)&pVgroup); pShow->pIter = sdbFetch(pSdb, SDB_VGROUP, pShow->pIter, (void **)&pVgroup);
if (pShow->pIter == NULL) break; if (pShow->pIter == NULL) break;
for (int32_t i = 0; i < pVgroup->replica && numOfRows < rows; ++i) { for (int32_t i = 0; i < pVgroup->replica && numOfRows < rows; ++i) {
SVnodeGid *pVgid = &pVgroup->vnodeGid[i]; SVnodeGid *pGid = &pVgroup->vnodeGid[i];
SColumnInfoData *pColInfo = NULL; SColumnInfoData *pColInfo = NULL;
cols = 0; cols = 0;
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataSetVal(pColInfo, numOfRows, (const char *)&pVgroup->vgId, false); colDataSetVal(pColInfo, numOfRows, (const char *)&pGid->dnodeId, false);
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataSetVal(pColInfo, numOfRows, (const char *)&pVgroup->replica, false);
char buf[20] = {0};
STR_TO_VARSTR(buf, syncStr(pVgid->syncState));
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataSetVal(pColInfo, numOfRows, (const char *)buf, false); colDataSetVal(pColInfo, numOfRows, (const char *)&pVgroup->vgId, false);
// db_name
const char *dbname = mndGetDbStr(pVgroup->dbName); const char *dbname = mndGetDbStr(pVgroup->dbName);
char b1[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0}; char b1[TSDB_DB_NAME_LEN + VARSTR_HEADER_SIZE] = {0};
if (dbname != NULL) { if (dbname != NULL) {
...@@ -992,20 +989,33 @@ static int32_t mndRetrieveVnodes(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pB ...@@ -992,20 +989,33 @@ static int32_t mndRetrieveVnodes(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pB
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataSetVal(pColInfo, numOfRows, (const char *)b1, false); colDataSetVal(pColInfo, numOfRows, (const char *)b1, false);
// dnode is online?
SDnodeObj *pDnode = mndAcquireDnode(pMnode, pGid->dnodeId);
if (pDnode == NULL) {
mError("failed to acquire dnode. dnodeId:%d", pGid->dnodeId);
break;
}
bool isDnodeOnline = mndIsDnodeOnline(pDnode, curMs);
char buf[20] = {0};
ESyncState syncState = (isDnodeOnline) ? pGid->syncState : TAOS_SYNC_STATE_OFFLINE;
STR_TO_VARSTR(buf, syncStr(syncState));
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataSetVal(pColInfo, numOfRows, (const char *)&pVgid->dnodeId, false); colDataSetVal(pColInfo, numOfRows, (const char *)buf, false);
int64_t roleTimeMs = (isDnodeOnline) ? pGid->roleTimeMs : 0;
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataSetVal(pColInfo, numOfRows, (const char *)&roleTimeMs, false);
int64_t startTimeMs = (isDnodeOnline) ? pGid->startTimeMs : 0;
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataSetVal(pColInfo, numOfRows, (const char *)&startTimeMs, false);
SDnodeObj *pDnode = mndAcquireDnode(pMnode, pVgid->dnodeId);
char b2[TSDB_EP_LEN + VARSTR_HEADER_SIZE] = {0};
if (pDnode != NULL) {
STR_WITH_MAXSIZE_TO_VARSTR(b2, pDnode->ep, TSDB_EP_LEN + VARSTR_HEADER_SIZE);
} else {
STR_WITH_MAXSIZE_TO_VARSTR(b2, "NULL", TSDB_EP_LEN + VARSTR_HEADER_SIZE);
}
pColInfo = taosArrayGet(pBlock->pDataBlock, cols++); pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);
colDataSetVal(pColInfo, numOfRows, (const char *)b2, false); colDataSetVal(pColInfo, numOfRows, (const char *)&pGid->syncRestore, false);
numOfRows++; numOfRows++;
sdbRelease(pSdb, pDnode);
} }
sdbRelease(pSdb, pVgroup); sdbRelease(pSdb, pVgroup);
......
...@@ -380,6 +380,9 @@ int32_t vnodeGetLoad(SVnode *pVnode, SVnodeLoad *pLoad) { ...@@ -380,6 +380,9 @@ int32_t vnodeGetLoad(SVnode *pVnode, SVnodeLoad *pLoad) {
pLoad->vgId = TD_VID(pVnode); pLoad->vgId = TD_VID(pVnode);
pLoad->syncState = state.state; pLoad->syncState = state.state;
pLoad->syncRestore = state.restored; pLoad->syncRestore = state.restored;
pLoad->syncTerm = state.term;
pLoad->roleTimeMs = state.roleTimeMs;
pLoad->startTimeMs = state.startTimeMs;
pLoad->syncCanRead = state.canRead; pLoad->syncCanRead = state.canRead;
pLoad->cacheUsage = tsdbCacheGetUsage(pVnode); pLoad->cacheUsage = tsdbCacheGetUsage(pVnode);
pLoad->numOfCachedTables = tsdbCacheGetElems(pVnode); pLoad->numOfCachedTables = tsdbCacheGetElems(pVnode);
......
...@@ -475,8 +475,8 @@ cmd ::= SHOW TAGS FROM table_name_cond(A) from_db_opt(B). ...@@ -475,8 +475,8 @@ cmd ::= SHOW TAGS FROM table_name_cond(A) from_db_opt(B).
cmd ::= SHOW TAGS FROM db_name(B) NK_DOT table_name(A). { pCxt->pRootNode = createShowStmtWithCond(pCxt, QUERY_NODE_SHOW_TAGS_STMT, createIdentifierValueNode(pCxt, &B), createIdentifierValueNode(pCxt, &A), OP_TYPE_EQUAL); } cmd ::= SHOW TAGS FROM db_name(B) NK_DOT table_name(A). { pCxt->pRootNode = createShowStmtWithCond(pCxt, QUERY_NODE_SHOW_TAGS_STMT, createIdentifierValueNode(pCxt, &B), createIdentifierValueNode(pCxt, &A), OP_TYPE_EQUAL); }
cmd ::= SHOW TABLE TAGS tag_list_opt(C) FROM table_name_cond(A) from_db_opt(B). { pCxt->pRootNode = createShowTableTagsStmt(pCxt, A, B, C); } cmd ::= SHOW TABLE TAGS tag_list_opt(C) FROM table_name_cond(A) from_db_opt(B). { pCxt->pRootNode = createShowTableTagsStmt(pCxt, A, B, C); }
cmd ::= SHOW TABLE TAGS tag_list_opt(C) FROM db_name(B) NK_DOT table_name(A). { pCxt->pRootNode = createShowTableTagsStmt(pCxt, createIdentifierValueNode(pCxt, &A), createIdentifierValueNode(pCxt, &B), C); } cmd ::= SHOW TABLE TAGS tag_list_opt(C) FROM db_name(B) NK_DOT table_name(A). { pCxt->pRootNode = createShowTableTagsStmt(pCxt, createIdentifierValueNode(pCxt, &A), createIdentifierValueNode(pCxt, &B), C); }
cmd ::= SHOW VNODES NK_INTEGER(A). { pCxt->pRootNode = createShowVnodesStmt(pCxt, createValueNode(pCxt, TSDB_DATA_TYPE_BIGINT, &A), NULL); } cmd ::= SHOW VNODES ON DNODE NK_INTEGER(A). { pCxt->pRootNode = createShowVnodesStmt(pCxt, createValueNode(pCxt, TSDB_DATA_TYPE_BIGINT, &A), NULL); }
cmd ::= SHOW VNODES NK_STRING(A). { pCxt->pRootNode = createShowVnodesStmt(pCxt, NULL, createValueNode(pCxt, TSDB_DATA_TYPE_VARCHAR, &A)); } cmd ::= SHOW VNODES. { pCxt->pRootNode = createShowVnodesStmt(pCxt, NULL, NULL); }
// show alive // show alive
cmd ::= SHOW db_name_cond_opt(A) ALIVE. { pCxt->pRootNode = createShowAliveStmt(pCxt, A, QUERY_NODE_SHOW_DB_ALIVE_STMT); } cmd ::= SHOW db_name_cond_opt(A) ALIVE. { pCxt->pRootNode = createShowAliveStmt(pCxt, A, QUERY_NODE_SHOW_DB_ALIVE_STMT); }
cmd ::= SHOW CLUSTER ALIVE. { pCxt->pRootNode = createShowAliveStmt(pCxt, NULL, QUERY_NODE_SHOW_CLUSTER_ALIVE_STMT); } cmd ::= SHOW CLUSTER ALIVE. { pCxt->pRootNode = createShowAliveStmt(pCxt, NULL, QUERY_NODE_SHOW_CLUSTER_ALIVE_STMT); }
......
...@@ -8073,8 +8073,6 @@ static int32_t rewriteShowVnodes(STranslateContext* pCxt, SQuery* pQuery) { ...@@ -8073,8 +8073,6 @@ static int32_t rewriteShowVnodes(STranslateContext* pCxt, SQuery* pQuery) {
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
if (NULL != pShow->pDnodeId) { if (NULL != pShow->pDnodeId) {
code = createOperatorNode(OP_TYPE_EQUAL, "dnode_id", pShow->pDnodeId, &pStmt->pWhere); code = createOperatorNode(OP_TYPE_EQUAL, "dnode_id", pShow->pDnodeId, &pStmt->pWhere);
} else {
code = createOperatorNode(OP_TYPE_EQUAL, "dnode_ep", pShow->pDnodeEndpoint, &pStmt->pWhere);
} }
} }
if (TSDB_CODE_SUCCESS == code) { if (TSDB_CODE_SUCCESS == code) {
......
此差异已折叠。
...@@ -239,9 +239,9 @@ TEST_F(ParserShowToUseTest, showVgroups) { ...@@ -239,9 +239,9 @@ TEST_F(ParserShowToUseTest, showVgroups) {
TEST_F(ParserShowToUseTest, showVnodes) { TEST_F(ParserShowToUseTest, showVnodes) {
useDb("root", "test"); useDb("root", "test");
run("SHOW VNODES 1"); run("SHOW VNODES ON DNODE 1");
run("SHOW VNODES 'node1:7030'"); run("SHOW VNODES");
} }
TEST_F(ParserShowToUseTest, splitVgroup) { TEST_F(ParserShowToUseTest, splitVgroup) {
......
...@@ -509,6 +509,7 @@ SSyncState syncGetState(int64_t rid) { ...@@ -509,6 +509,7 @@ SSyncState syncGetState(int64_t rid) {
if (pSyncNode != NULL) { if (pSyncNode != NULL) {
state.state = pSyncNode->state; state.state = pSyncNode->state;
state.roleTimeMs = pSyncNode->roleTimeMs; state.roleTimeMs = pSyncNode->roleTimeMs;
state.startTimeMs = pSyncNode->startTime;
state.restored = pSyncNode->restoreFinish; state.restored = pSyncNode->restoreFinish;
if (pSyncNode->vgId != 1) { if (pSyncNode->vgId != 1) {
state.canRead = syncNodeIsReadyForRead(pSyncNode); state.canRead = syncNodeIsReadyForRead(pSyncNode);
......
...@@ -107,35 +107,39 @@ if $data30 != 12 then ...@@ -107,35 +107,39 @@ if $data30 != 12 then
return -1 return -1
endi endi
print =============== show vnodes print =============== show vnodes on dnode 1
sql show vnodes 1 sql show vnodes on dnode 1
if $rows != 9 then if $rows != 9 then
return -1 return -1
endi endi
if $data(4)[1] != 1 then if $data10 != 1 then
return -1 return -1
endi endi
if $data(4)[2] != leader then if $data11 != 5 then
return -1 return -1
endi endi
if $data(4)[3] != d2 then if $data12 != d2 then
return -1 return -1
endi endi
if $data(4)[4] != 1 then if $data13 != leader then
return -1 return -1
endi endi
if $data(4)[5] != localhost:7100 then print $data14
return -1 print $data15
if $data16 != 1 then
return -1
endi endi
sql show vnodes 'localhost:7100' print ================ show vnodes
sql show vnodes
if $rows != 9 then if $rows != 9 then
return -1 return -1
endi endi
print =============== drop database print =============== drop database
......
...@@ -128,7 +128,7 @@ class TDTestCase: ...@@ -128,7 +128,7 @@ class TDTestCase:
continue continue
else: else:
tdLog.exit(f"show create database check failed with {key} {value}") tdLog.exit(f"show create database check failed with {key} {value}")
tdSql.query('show vnodes 1') tdSql.query('show vnodes on dnode 1')
tdSql.checkRows(self.vgroups) tdSql.checkRows(self.vgroups)
tdSql.execute(f'use {self.dbname}') tdSql.execute(f'use {self.dbname}')
......
Markdown is supported
0% .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑!
想要评论请 注册