From 2bf71e71079ed55d68aecad4040afcde1914313f Mon Sep 17 00:00:00 2001 From: slguan Date: Tue, 10 Dec 2019 13:00:00 +0800 Subject: [PATCH] [TBASE-1241] --- src/client/src/tscServer.c | 5 +++ src/client/src/tscSql.c | 32 ++------------- src/inc/taosmsg.h | 4 ++ src/inc/tglobalcfg.h | 2 + src/inc/tutil.h | 2 + src/system/detail/inc/mgmt.h | 4 +- src/system/detail/src/dnodeService.c | 8 +--- src/system/detail/src/mgmtMeter.c | 22 ++++++---- src/system/detail/src/mgmtShell.c | 61 ++++++++++++++++++++++------ src/system/detail/src/mgmtVgroup.c | 16 +++++--- src/util/src/tglobalcfg.c | 10 +++++ src/util/src/tutil.c | 36 ++++++++++++++-- 12 files changed, 136 insertions(+), 66 deletions(-) diff --git a/src/client/src/tscServer.c b/src/client/src/tscServer.c index d25b2dde53..3fac0c74f2 100644 --- a/src/client/src/tscServer.c +++ b/src/client/src/tscServer.c @@ -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; diff --git a/src/client/src/tscSql.c b/src/client/src/tscSql.c index 233a37dc59..381b342fdc 100644 --- a/src/client/src/tscSql.c +++ b/src/client/src/tscSql.c @@ -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; diff --git a/src/inc/taosmsg.h b/src/inc/taosmsg.h index c1820a5b9c..ae58c43477 100644 --- a/src/inc/taosmsg.h +++ b/src/inc/taosmsg.h @@ -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 { diff --git a/src/inc/tglobalcfg.h b/src/inc/tglobalcfg.h index 99bc3431d6..0bf47d75ad 100644 --- a/src/inc/tglobalcfg.h +++ b/src/inc/tglobalcfg.h @@ -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[]; diff --git a/src/inc/tutil.h b/src/inc/tutil.h index bdf9df63ee..df82c2113a 100644 --- a/src/inc/tutil.h +++ b/src/inc/tutil.h @@ -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 diff --git a/src/system/detail/inc/mgmt.h b/src/system/detail/inc/mgmt.h index 75ec841076..b59c5cf61c 100644 --- a/src/system/detail/inc/mgmt.h +++ b/src/system/detail/inc/mgmt.h @@ -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); diff --git a/src/system/detail/src/dnodeService.c b/src/system/detail/src/dnodeService.c index f03bd5f3bb..eec918acd0 100644 --- a/src/system/detail/src/dnodeService.c +++ b/src/system/detail/src/dnodeService.c @@ -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; diff --git a/src/system/detail/src/mgmtMeter.c b/src/system/detail/src/mgmtMeter.c index 4e1d62e60e..9acd4f1326 100644 --- a/src/system/detail/src/mgmtMeter.c +++ b/src/system/detail/src/mgmtMeter.c @@ -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]); diff --git a/src/system/detail/src/mgmtShell.c b/src/system/detail/src/mgmtShell.c index e6afe11855..576353d758 100644 --- a/src/system/detail/src/mgmtShell.c +++ b/src/system/detail/src/mgmtShell.c @@ -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 diff --git a/src/system/detail/src/mgmtVgroup.c b/src/system/detail/src/mgmtVgroup.c index e3bed57b33..d7e935241a 100644 --- a/src/system/detail/src/mgmtVgroup.c +++ b/src/system/detail/src/mgmtVgroup.c @@ -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); } diff --git a/src/util/src/tglobalcfg.c b/src/util/src/tglobalcfg.c index aabc722519..345dbdd4db 100644 --- a/src/util/src/tglobalcfg.c +++ b/src/util/src/tglobalcfg.c @@ -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, diff --git a/src/util/src/tutil.c b/src/util/src/tutil.c index e0dcb0aa3f..f21ca96a81 100644 --- a/src/util/src/tutil.c +++ b/src/util/src/tutil.c @@ -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;