xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

[TBASE-1241]

This commit is contained in:
slguan 2019-12-10 13:00:00 +08:00
parent fe4fb35392
commit 2bf71e7107
12 changed files with 136 additions and 66 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
src/client/src/tscServer.c
View File

@ -2642,6 +2642,11 @@ int tscBuildConnectMsg(SSqlObj *pSql) {
db = (db == NULL) ? pObj->db : db + 1;
strcpy(pConnect->db, db);
strcpy(pConnect->clientVersion, version);
pConnect->usePublicIp = (int8_t)tsUsePublicIp;
pConnect->isCluster = (int8_t)tsIsCluster;
memset(pConnect->reserved, 0, sizeof(pConnect->reserved));
pMsg += sizeof(SConnectMsg);
msgLen = pMsg - pStart;

32
src/client/src/tscSql.c
View File

@ -165,38 +165,12 @@ TAOS *taos_connect(const char *ip, const char *user, const char *pass, const cha
STscObj* pObj = (STscObj*) taos;
// version compare only requires the first 3 segments of the version string
int32_t comparedSegments = 3;
char client_version[64] = {0};
char server_version[64] = {0};
int clientVersionNumber[4] = {0};
int serverVersionNumber[4] = {0};
strcpy(client_version, version);
strcpy(server_version, taos_get_server_info(taos));
if (!taosGetVersionNumber(client_version, clientVersionNumber)) {
tscError("taos:%p, invalid client version:%s", taos, client_version);
pObj->pSql->res.code = TSDB_CODE_INVALID_CLIENT_VERSION;
int code = taosCheckVersion(version, taos_get_server_info(taos), 3);
if (code != 0) {
pObj->pSql->res.code = code;
taos_close(taos);
return NULL;
}
if (!taosGetVersionNumber(server_version, serverVersionNumber)) {
tscError("taos:%p, invalid server version:%s", taos, server_version);
pObj->pSql->res.code = TSDB_CODE_INVALID_CLIENT_VERSION;
taos_close(taos);
return NULL;
}
for(int32_t i = 0; i < comparedSegments; ++i) {
if (clientVersionNumber[i] != serverVersionNumber[i]) {
tscError("taos:%p, the %d-th number of server version:%s not matched with client version:%s, close connection",
taos, i, server_version, version);
pObj->pSql->res.code = TSDB_CODE_INVALID_CLIENT_VERSION;
taos_close(taos);
return NULL;
}
}
}
return taos;

4
src/inc/taosmsg.h
View File

@ -350,7 +350,11 @@ typedef struct {
} SAlterTableMsg;
typedef struct {
char clientVersion[TSDB_VERSION_LEN];
char db[TSDB_METER_ID_LEN];
int8_t usePublicIp;
int8_t isCluster;
int8_t reserved[14];
} SConnectMsg;
typedef struct {

2
src/inc/tglobalcfg.h
View File

@ -151,6 +151,8 @@ extern int tsAdminRowLimit;
extern int tsTscEnableRecordSql;
extern int tsAnyIp;
extern int tsUsePublicIp;
extern int tsIsCluster;
extern char tsMonitorDbName[];
extern char tsInternalPass[];

2
src/inc/tutil.h
View File

@ -187,6 +187,8 @@ static FORCE_INLINE void taosEncryptPass(uint8_t *inBuf, unsigned int inLen, cha
memcpy(target, context.digest, TSDB_KEY_LEN);
}
int taosCheckVersion(char *input_client_version, char *input_server_version, int compared_segments);
char *taosIpStr(uint32_t ipInt);
#define TAOS_ALLOC_MODE_DEFAULT 0

4
src/system/detail/inc/mgmt.h
View File

@ -222,6 +222,8 @@ typedef struct _connObj {
char superAuth : 1; // super user flag
char writeAuth : 1; // write flag
char killConnection : 1; // kill the connection flag
char usePublicIp : 1; // if the connection request is publicIp
char reserved : 4;
uint32_t queryId; // query ID to be killed
uint32_t streamId; // stream ID to be killed
uint32_t ip; // shell IP
@ -343,7 +345,7 @@ void mgmtCleanUpVgroups();
int mgmtInitMeters();
STabObj *mgmtGetMeter(char *meterId);
STabObj *mgmtGetMeterInfo(char *src, char *tags[]);
int mgmtRetrieveMetricMeta(void *thandle, char **pStart, SMetricMetaMsg *pInfo);
int mgmtRetrieveMetricMeta(SConnObj *pConn, char **pStart, SMetricMetaMsg *pInfo);
int mgmtCreateMeter(SDbObj *pDb, SCreateTableMsg *pCreate);
int mgmtDropMeter(SDbObj *pDb, char *meterId, int ignore);
int mgmtAlterMeter(SDbObj *pDb, SAlterTableMsg *pAlter);

8
src/system/detail/src/dnodeService.c
View File

@ -55,12 +55,8 @@ int main(int argc, char *argv[]) {
exit(EXIT_FAILURE);
}
} else if (strcmp(argv[i], "-V") == 0) {
#ifdef CLUSTER
printf("enterprise version: %s compatible_version: %s\n", version, compatible_version);
#else
printf("community version: %s compatible_version: %s\n", version, compatible_version);
#endif
char *versionStr = tsIsCluster ? "enterprise" : "community";
printf("%s version: %s compatible_version: %s\n", versionStr, version, compatible_version);
printf("gitinfo: %s\n", gitinfo);
printf("buildinfo: %s\n", buildinfo);
return 0;

22
src/system/detail/src/mgmtMeter.c
View File

@ -987,7 +987,7 @@ SSchema *mgmtGetMeterSchema(STabObj *pMeter) {
/*
* serialize SVnodeSidList to byte array
*/
static char *mgmtBuildMetricMetaMsg(STabObj *pMeter, int32_t *ovgId, SVnodeSidList **pList, SMetricMeta *pMeta,
static char *mgmtBuildMetricMetaMsg(SConnObj *pConn, STabObj *pMeter, int32_t *ovgId, SVnodeSidList **pList, SMetricMeta *pMeta,
int32_t tagLen, int16_t numOfTags, int16_t *tagsId, int32_t maxNumOfMeters,
char *pMsg) {
if (pMeter->gid.vgId != *ovgId || ((*pList) != NULL && (*pList)->numOfSids >= maxNumOfMeters)) {
@ -1004,8 +1004,13 @@ static char *mgmtBuildMetricMetaMsg(STabObj *pMeter, int32_t *ovgId, SVnodeSidLi
SVgObj *pVgroup = mgmtGetVgroup(pMeter->gid.vgId);
for (int i = 0; i < TSDB_VNODES_SUPPORT; ++i) {
(*pList)->vpeerDesc[i].ip = pVgroup->vnodeGid[i].publicIp;
(*pList)->vpeerDesc[i].vnode = pVgroup->vnodeGid[i].vnode;
if (pConn->usePublicIp) {
(*pList)->vpeerDesc[i].ip = pVgroup->vnodeGid[i].publicIp;
(*pList)->vpeerDesc[i].vnode = pVgroup->vnodeGid[i].vnode;
} else {
(*pList)->vpeerDesc[i].ip = pVgroup->vnodeGid[i].ip;
(*pList)->vpeerDesc[i].vnode = pVgroup->vnodeGid[i].vnode;
}
}
pMsg += sizeof(SVnodeSidList);
@ -1102,10 +1107,10 @@ static SMetricMetaElemMsg *doConvertMetricMetaMsg(SMetricMetaMsg *pMetricMetaMsg
return pElem;
}
static int32_t mgmtBuildMetricMetaRspMsg(void *thandle, SMetricMetaMsg *pMetricMetaMsg, tQueryResultset *pResult,
static int32_t mgmtBuildMetricMetaRspMsg(SConnObj *pConn, SMetricMetaMsg *pMetricMetaMsg, tQueryResultset *pResult,
char **pStart, int32_t *tagLen, int32_t rspMsgSize, int32_t maxTablePerVnode,
int32_t code) {
*pStart = taosBuildRspMsgWithSize(thandle, TSDB_MSG_TYPE_METRIC_META_RSP, rspMsgSize);
*pStart = taosBuildRspMsgWithSize(pConn->thandle, TSDB_MSG_TYPE_METRIC_META_RSP, rspMsgSize);
if (*pStart == NULL) {
return 0;
}
@ -1143,7 +1148,7 @@ static int32_t mgmtBuildMetricMetaRspMsg(void *thandle, SMetricMetaMsg *pMetricM
for (int32_t i = 0; i < pResult[j].num; ++i) {
STabObj *pMeter = pResult[j].pRes[i];
pMsg = mgmtBuildMetricMetaMsg(pMeter, &ovgId, &pList, pMeta, tagLen[j], pElem->numOfTags, pElem->tagCols,
pMsg = mgmtBuildMetricMetaMsg(pConn, pMeter, &ovgId, &pList, pMeta, tagLen[j], pElem->numOfTags, pElem->tagCols,
maxTablePerVnode, pMsg);
}
@ -1159,7 +1164,7 @@ static int32_t mgmtBuildMetricMetaRspMsg(void *thandle, SMetricMetaMsg *pMetricM
return msgLen;
}
int mgmtRetrieveMetricMeta(void *thandle, char **pStart, SMetricMetaMsg *pMetricMetaMsg) {
int mgmtRetrieveMetricMeta(SConnObj *pConn, char **pStart, SMetricMetaMsg *pMetricMetaMsg) {
/*
* naive method: Do not limit the maximum number of meters in each
* vnode(subquery), split the result according to vnodes
@ -1233,8 +1238,7 @@ int mgmtRetrieveMetricMeta(void *thandle, char **pStart, SMetricMetaMsg *pMetric
msgLen = 512;
}
msgLen = mgmtBuildMetricMetaRspMsg(thandle, pMetricMetaMsg, result, pStart, tagLen, msgLen, maxMetersPerVNodeForQuery,
ret);
msgLen = mgmtBuildMetricMetaRspMsg(pConn, pMetricMetaMsg, result, pStart, tagLen, msgLen, maxMetersPerVNodeForQuery, ret);
for (int32_t i = 0; i < pMetricMetaMsg->numOfMeters; ++i) {
tQueryResultClean(&result[i]);

61
src/system/detail/src/mgmtShell.c
View File

@ -314,8 +314,13 @@ int mgmtProcessMeterMetaMsg(char *pMsg, int msgLen, SConnObj *pConn) {
goto _exit_code;
}
for (int i = 0; i < TSDB_VNODES_SUPPORT; ++i) {
pMeta->vpeerDesc[i].ip = pVgroup->vnodeGid[i].publicIp;
pMeta->vpeerDesc[i].vnode = htonl(pVgroup->vnodeGid[i].vnode);
if (pConn->usePublicIp) {
pMeta->vpeerDesc[i].ip = pVgroup->vnodeGid[i].publicIp;
pMeta->vpeerDesc[i].vnode = htonl(pVgroup->vnodeGid[i].vnode);
} else {
pMeta->vpeerDesc[i].ip = pVgroup->vnodeGid[i].ip;
pMeta->vpeerDesc[i].vnode = htonl(pVgroup->vnodeGid[i].vnode);
}
}
}
}
@ -453,8 +458,13 @@ int mgmtProcessMultiMeterMetaMsg(char *pMsg, int msgLen, SConnObj *pConn) {
}
for (int i = 0; i < TSDB_VNODES_SUPPORT; ++i) {
pMeta->meta.vpeerDesc[i].ip = pVgroup->vnodeGid[i].publicIp;
pMeta->meta.vpeerDesc[i].vnode = htonl(pVgroup->vnodeGid[i].vnode);
if (pConn->usePublicIp) {
pMeta->meta.vpeerDesc[i].ip = pVgroup->vnodeGid[i].publicIp;
pMeta->meta.vpeerDesc[i].vnode = htonl(pVgroup->vnodeGid[i].vnode);
} else {
pMeta->meta.vpeerDesc[i].ip = pVgroup->vnodeGid[i].ip;
pMeta->meta.vpeerDesc[i].vnode = htonl(pVgroup->vnodeGid[i].vnode);
}
}
}
}
@ -526,7 +536,7 @@ int mgmtProcessMetricMetaMsg(char *pMsg, int msgLen, SConnObj *pConn) {
msgLen = pMsg - pStart;
} else {
msgLen = mgmtRetrieveMetricMeta(pConn->thandle, &pStart, pMetricMetaMsg);
msgLen = mgmtRetrieveMetricMeta(pConn, &pStart, pMetricMetaMsg);
if (msgLen <= 0) {
taosSendSimpleRsp(pConn->thandle, TSDB_MSG_TYPE_METRIC_META_RSP, TSDB_CODE_SERV_OUT_OF_MEMORY);
return 0;
@ -1099,10 +1109,17 @@ int mgmtProcessHeartBeatMsg(char *cont, int contLen, SConnObj *pConn) {
pHBRsp->killConnection = pConn->killConnection;
#ifdef CLUSTER
int size = pSdbPublicIpList->numOfIps * 4;
pHBRsp->ipList.numOfIps = pSdbPublicIpList->numOfIps;
memcpy(pHBRsp->ipList.ip, pSdbPublicIpList->ip, size);
pMsg += sizeof(SHeartBeatRsp) + size;
if (pConn->usePublicIp) {
int size = pSdbPublicIpList->numOfIps * 4;
pHBRsp->ipList.numOfIps = pSdbPublicIpList->numOfIps;
memcpy(pHBRsp->ipList.ip, pSdbPublicIpList->ip, size);
pMsg += sizeof(SHeartBeatRsp) + size;
} else {
int size = pSdbIpList->numOfIps * 4;
pHBRsp->ipList.numOfIps = pSdbIpList->numOfIps;
memcpy(pHBRsp->ipList.ip, pSdbIpList->ip, size);
pMsg += sizeof(SHeartBeatRsp) + size;
}
#else
pMsg += sizeof(SHeartBeatRsp);
#endif
@ -1178,6 +1195,18 @@ int mgmtProcessConnectMsg(char *pMsg, int msgLen, SConnObj *pConn) {
pAcct = mgmtGetAcct(pUser->acct);
code = taosCheckVersion(pConnectMsg->clientVersion, version, 3);
if (code != 0) {
mError("invalid client version:%s", pConnectMsg->clientVersion);
goto _rsp;
}
if (pConnectMsg->isCluster != tsIsCluster) {
mError("Cluster Edition and lite Edition cannot be interconnected, client:%d server:%d", pConnectMsg->isCluster, tsIsCluster);
code = TSDB_CODE_INVALID_CLIENT_VERSION;
goto _rsp;
}
if (pConnectMsg->db[0]) {
sprintf(dbName, "%x%s%s", pAcct->acctId, TS_PATH_DELIMITER, pConnectMsg->db);
pDb = mgmtGetDb(dbName);
@ -1217,9 +1246,17 @@ _rsp:
pMsg += sizeof(SConnectRsp);
#ifdef CLUSTER
int size = pSdbPublicIpList->numOfIps * 4 + sizeof(SIpList);
memcpy(pMsg, pSdbPublicIpList, size);
pMsg += size;
if (pConnectMsg->usePublicIp) {
pConn->usePublicIp = 1;
int size = pSdbPublicIpList->numOfIps * 4 + sizeof(SIpList);
memcpy(pMsg, pSdbPublicIpList, size);
pMsg += size;
}
else {
int size = pSdbIpList->numOfIps * 4 + sizeof(SIpList);
memcpy(pMsg, pSdbIpList, size);
pMsg += size;
}
#endif
// set the time resolution: millisecond or microsecond

16
src/system/detail/src/mgmtVgroup.c
View File

@ -102,13 +102,17 @@ int mgmtInitVgroups() {
}
taosIdPoolReinit(pVgroup->idPool);
#ifdef CLUSTER
if (pVgroup->vnodeGid[0].publicIp == 0) {
pVgroup->vnodeGid[0].publicIp = inet_addr(tsPublicIp);
pVgroup->vnodeGid[0].ip = inet_addr(tsPrivateIp);
sdbUpdateRow(vgSdb, pVgroup, tsVgUpdateSize, 1);
if (tsIsCluster) {
/*
* Upgrade from open source version to cluster version for the first time
*/
if (pVgroup->vnodeGid[0].publicIp == 0) {
pVgroup->vnodeGid[0].publicIp = inet_addr(tsPublicIp);
pVgroup->vnodeGid[0].ip = inet_addr(tsPrivateIp);
sdbUpdateRow(vgSdb, pVgroup, tsVgUpdateSize, 1);
}
}
#endif
mgmtSetDnodeVgid(pVgroup->vnodeGid, pVgroup->numOfVnodes, pVgroup->vgId);
}

10
src/util/src/tglobalcfg.c
View File

@ -164,6 +164,13 @@ int tsAdminRowLimit = 10240;
int tsTscEnableRecordSql = 0;
int tsEnableCoreFile = 0;
int tsAnyIp = 1;
int tsUsePublicIp = 0;
#ifdef CLUSTER
int tsIsCluster = 1;
#else
int tsIsCluster = 0;
#endif
int tsRpcTimer = 300;
int tsRpcMaxTime = 600; // seconds;
@ -780,6 +787,9 @@ static void doInitGlobalConfig() {
TSDB_CFG_CTYPE_B_CONFIG,
0, 1, 0, TSDB_CFG_UTYPE_NONE);
tsInitConfigOption(cfg++, "usePublicIp", &tsUsePublicIp, TSDB_CFG_VTYPE_INT,
TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_CLIENT,
0, 1, 0, TSDB_CFG_UTYPE_NONE);
// version info
tsInitConfigOption(cfg++, "gitinfo", gitinfo, TSDB_CFG_VTYPE_STRING,
TSDB_CFG_CTYPE_B_SHOW | TSDB_CFG_CTYPE_B_CLIENT,

36
src/util/src/tutil.c
View File

@ -24,6 +24,8 @@
#include "ttime.h"
#include "ttypes.h"
#include "tutil.h"
#include "tlog.h"
#include "taoserror.h"
int32_t strdequote(char *z) {
if (z == NULL) {
@ -450,10 +452,8 @@ bool taosValidateEncodec(char *encodec) {
return false;
}
iconv_close(cd);
return true;
#else
return true;
#endif
return true;
}
bool taosGetVersionNumber(char *versionStr, int *versionNubmer) {
@ -485,6 +485,36 @@ bool taosGetVersionNumber(char *versionStr, int *versionNubmer) {
return true;
}
int taosCheckVersion(char *input_client_version, char *input_server_version, int comparedSegments) {
char client_version[64] = {0};
char server_version[64] = {0};
int clientVersionNumber[4] = {0};
int serverVersionNumber[4] = {0};
strcpy(client_version, input_client_version);
strcpy(server_version, input_server_version);
if (!taosGetVersionNumber(client_version, clientVersionNumber)) {
pError("invalid client version:%s", client_version);
return TSDB_CODE_INVALID_CLIENT_VERSION;
}
if (!taosGetVersionNumber(server_version, serverVersionNumber)) {
pError("invalid server version:%s", server_version);
return TSDB_CODE_INVALID_CLIENT_VERSION;
return NULL;
}
for(int32_t i = 0; i < comparedSegments; ++i) {
if (clientVersionNumber[i] != serverVersionNumber[i]) {
tscError("the %d-th number of server version:%s not matched with client version:%s", i, server_version, version);
return TSDB_CODE_INVALID_CLIENT_VERSION;
}
}
return 0;
}
char *taosIpStr(uint32_t ipInt) {
static char ipStrArray[3][30];
static int ipStrIndex = 0;