Skip to content

T
TDengine

慢慢CG / TDengine
与 Fork 源项目一致

Fork自 taosdata / TDengine

通知 1
Star 0
Fork 0
T
TDengine
“83d516af4cce21e4f1284f65a5ba5d29a0c28b42”上不存在“source/dnode/mgmt/qnode/inc/qm.h”
未验证 提交 be2c6f5d 编写于 作者: S slguan 提交者： GitHub
浏览文件
操作

Merge pull request #905 from taosdata/hotfix/TBASE-1241 

Changed publicIp communication method
上级 59fd6fe8 f8453c14
显示空白变更内容
内联 并排
Showing with 145 addition and 111 deletion
src/client/src/tscServer.c
浏览文件 @ be2c6f5d
...@@ -2650,6 +2650,8 @@ int tscBuildConnectMsg(SSqlObj *pSql) { ...@@ -2650,6 +2650,8 @@ int tscBuildConnectMsg(SSqlObj *pSql) {
db = (db == NULL) ? pObj->db : db + 1; db = (db == NULL) ? pObj->db : db + 1;
strcpy(pConnect->db, db); strcpy(pConnect->db, db);
strcpy(pConnect->clientVersion, version);
pMsg += sizeof(SConnectMsg); pMsg += sizeof(SConnectMsg);
msgLen = pMsg - pStart; msgLen = pMsg - pStart;
......
src/client/src/tscSql.c
浏览文件 @ be2c6f5d
...@@ -165,38 +165,12 @@ TAOS *taos_connect(const char *ip, const char *user, const char *pass, const cha ...@@ -165,38 +165,12 @@ TAOS *taos_connect(const char *ip, const char *user, const char *pass, const cha
STscObj *pObj = (STscObj *)taos; STscObj *pObj = (STscObj *)taos;
// version compare only requires the first 3 segments of the version string // version compare only requires the first 3 segments of the version string
int32_t comparedSegments = 3; int code = taosCheckVersion(version, taos_get_server_info(taos), 3);
char client_version[64] = {0}; if (code != 0) {
char server_version[64] = {0}; pObj->pSql->res.code = code;
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;
taos_close(taos); taos_close(taos);
return NULL; 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; return taos;
......
src/inc/taosmsg.h
浏览文件 @ be2c6f5d
...@@ -222,6 +222,7 @@ typedef struct { ...@@ -222,6 +222,7 @@ typedef struct {
// internal part // internal part
uint32_t destId; uint32_t destId;
uint32_t destIp;
char meterId[TSDB_UNI_LEN]; char meterId[TSDB_UNI_LEN];
uint16_t port; // for UDP only uint16_t port; // for UDP only
char empty[1]; char empty[1];
...@@ -350,6 +351,7 @@ typedef struct { ...@@ -350,6 +351,7 @@ typedef struct {
} SAlterTableMsg; } SAlterTableMsg;
typedef struct { typedef struct {
char clientVersion[TSDB_VERSION_LEN];
char db[TSDB_METER_ID_LEN]; char db[TSDB_METER_ID_LEN];
} SConnectMsg; } SConnectMsg;
...@@ -662,6 +664,7 @@ typedef struct { ...@@ -662,6 +664,7 @@ typedef struct {
// internal message // internal message
typedef struct { typedef struct {
uint32_t destId; uint32_t destId;
uint32_t destIp;
char meterId[TSDB_UNI_LEN]; char meterId[TSDB_UNI_LEN];
char empty[3]; char empty[3];
uint8_t msgType; uint8_t msgType;
......
src/inc/tglobalcfg.h
浏览文件 @ be2c6f5d
...@@ -74,13 +74,13 @@ extern int tsMetricMetaKeepTimer; ...@@ -74,13 +74,13 @@ extern int tsMetricMetaKeepTimer;
extern float tsNumOfThreadsPerCore; extern float tsNumOfThreadsPerCore;
extern float tsRatioOfQueryThreads; extern float tsRatioOfQueryThreads;
extern char tsPublicIp[]; extern char tsPublicIp[];
extern char tsInternalIp[];
extern char tsPrivateIp[]; extern char tsPrivateIp[];
extern char tsServerIpStr[]; extern char tsServerIpStr[];
extern short tsNumOfVnodesPerCore; extern short tsNumOfVnodesPerCore;
extern short tsNumOfTotalVnodes; extern short tsNumOfTotalVnodes;
extern short tsCheckHeaderFile; extern short tsCheckHeaderFile;
extern uint32_t tsServerIp; extern uint32_t tsServerIp;
extern uint32_t tsPublicIpInt;
extern int tsSessionsPerVnode; extern int tsSessionsPerVnode;
extern int tsAverageCacheBlocks; extern int tsAverageCacheBlocks;
...@@ -151,6 +151,8 @@ extern int tsTelegrafUseFieldNum; ...@@ -151,6 +151,8 @@ extern int tsTelegrafUseFieldNum;
extern int tsAdminRowLimit; extern int tsAdminRowLimit;
extern int tsTscEnableRecordSql; extern int tsTscEnableRecordSql;
extern int tsAnyIp;
extern int tsIsCluster;
extern char tsMonitorDbName[]; extern char tsMonitorDbName[];
extern char tsInternalPass[]; extern char tsInternalPass[];
......
src/inc/tutil.h
浏览文件 @ be2c6f5d
...@@ -187,6 +187,8 @@ static FORCE_INLINE void taosEncryptPass(uint8_t *inBuf, unsigned int inLen, cha ...@@ -187,6 +187,8 @@ static FORCE_INLINE void taosEncryptPass(uint8_t *inBuf, unsigned int inLen, cha
memcpy(target, context.digest, TSDB_KEY_LEN); 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); char *taosIpStr(uint32_t ipInt);
uint32_t ip2uint(const char *const ip_addr); uint32_t ip2uint(const char *const ip_addr);
......
src/modules/monitor/src/monitorSystem.c
浏览文件 @ be2c6f5d
...@@ -114,11 +114,7 @@ void monitorInitConn(void *para, void *unused) { ...@@ -114,11 +114,7 @@ void monitorInitConn(void *para, void *unused) {
monitor->state = MONITOR_STATE_INITIALIZING; monitor->state = MONITOR_STATE_INITIALIZING;
if (monitor->privateIpStr[0] == 0) { if (monitor->privateIpStr[0] == 0) {
#ifdef CLUSTER
strcpy(monitor->privateIpStr, tsPrivateIp); strcpy(monitor->privateIpStr, tsPrivateIp);
#else
strcpy(monitor->privateIpStr, tsInternalIp);
#endif
for (int i = 0; i < TSDB_IPv4ADDR_LEN; ++i) { for (int i = 0; i < TSDB_IPv4ADDR_LEN; ++i) {
if (monitor->privateIpStr[i] == '.') { if (monitor->privateIpStr[i] == '.') {
monitor->privateIpStr[i] = '_'; monitor->privateIpStr[i] = '_';
...@@ -168,11 +164,7 @@ void dnodeBuildMonitorSql(char *sql, int cmd) { ...@@ -168,11 +164,7 @@ void dnodeBuildMonitorSql(char *sql, int cmd) {
tsMonitorDbName, IP_LEN_STR + 1); tsMonitorDbName, IP_LEN_STR + 1);
} else if (cmd == MONITOR_CMD_CREATE_TB_DN) { } else if (cmd == MONITOR_CMD_CREATE_TB_DN) {
snprintf(sql, SQL_LENGTH, "create table if not exists %s.dn_%s using %s.dn tags('%s')", tsMonitorDbName, snprintf(sql, SQL_LENGTH, "create table if not exists %s.dn_%s using %s.dn tags('%s')", tsMonitorDbName,
#ifdef CLUSTER
monitor->privateIpStr, tsMonitorDbName, tsPrivateIp); monitor->privateIpStr, tsMonitorDbName, tsPrivateIp);
#else
monitor->privateIpStr, tsMonitorDbName, tsInternalIp);
#endif
} else if (cmd == MONITOR_CMD_CREATE_MT_ACCT) { } else if (cmd == MONITOR_CMD_CREATE_MT_ACCT) {
snprintf(sql, SQL_LENGTH, snprintf(sql, SQL_LENGTH,
"create table if not exists %s.acct(ts timestamp " "create table if not exists %s.acct(ts timestamp "
...@@ -227,10 +219,8 @@ void monitorInitDatabaseCb(void *param, TAOS_RES *result, int code) { ...@@ -227,10 +219,8 @@ void monitorInitDatabaseCb(void *param, TAOS_RES *result, int code) {
taosLogSqlFp = monitorExecuteSQL; taosLogSqlFp = monitorExecuteSQL;
#ifdef CLUSTER #ifdef CLUSTER
taosLogAcctFp = monitorSaveAcctLog; taosLogAcctFp = monitorSaveAcctLog;
monitorLPrint("dnode:%s is started", tsPrivateIp);
#else
monitorLPrint("dnode:%s is started", tsInternalIp);
#endif #endif
monitorLPrint("dnode:%s is started", tsPrivateIp);
} }
monitor->cmdIndex++; monitor->cmdIndex++;
monitorInitDatabase(); monitorInitDatabase();
...@@ -246,11 +236,7 @@ void monitorStopSystem() { ...@@ -246,11 +236,7 @@ void monitorStopSystem() {
return; return;
} }
#ifdef CLUSTER
monitorLPrint("dnode:%s monitor module is stopped", tsPrivateIp); monitorLPrint("dnode:%s monitor module is stopped", tsPrivateIp);
#else
monitorLPrint("dnode:%s monitor module is stopped", tsInternalIp);
#endif
monitor->state = MONITOR_STATE_STOPPED; monitor->state = MONITOR_STATE_STOPPED;
taosLogFp = NULL; taosLogFp = NULL;
if (monitor->initTimer != NULL) { if (monitor->initTimer != NULL) {
...@@ -440,11 +426,7 @@ void monitorSaveLog(int level, const char *const format, ...) { ...@@ -440,11 +426,7 @@ void monitorSaveLog(int level, const char *const format, ...) {
va_end(argpointer); va_end(argpointer);
if (len > max_length) len = max_length; if (len > max_length) len = max_length;
#ifdef CLUSTER
len += sprintf(sql + len, "', '%s')", tsPrivateIp); len += sprintf(sql + len, "', '%s')", tsPrivateIp);
#else
len += sprintf(sql + len, "', '%s')", tsInternalIp);
#endif
sql[len++] = 0; sql[len++] = 0;
monitorTrace("monitor:%p, save log, sql: %s", monitor->conn, sql); monitorTrace("monitor:%p, save log, sql: %s", monitor->conn, sql);
......
src/os/linux/src/tsystem.c
浏览文件 @ be2c6f5d
...@@ -381,8 +381,8 @@ bool taosGetCardName(char *ip, char *name) { ...@@ -381,8 +381,8 @@ bool taosGetCardName(char *ip, char *name) {
bool taosGetCardInfo(int64_t *bytes) { bool taosGetCardInfo(int64_t *bytes) {
static char tsPublicCard[1000] = {0}; static char tsPublicCard[1000] = {0};
if (tsPublicCard[0] == 0) { if (tsPublicCard[0] == 0) {
if (!taosGetCardName(tsInternalIp, tsPublicCard)) { if (!taosGetCardName(tsPrivateIp, tsPublicCard)) {
pError("can't get card name from ip:%s", tsInternalIp); pError("can't get card name from ip:%s", tsPrivateIp);
return false; return false;
} }
int cardNameLen = (int)strlen(tsPublicCard); int cardNameLen = (int)strlen(tsPublicCard);
......
src/rpc/src/trpc.c
浏览文件 @ be2c6f5d
...@@ -1222,6 +1222,7 @@ int taosSendMsgToPeerH(void *thandle, char *pCont, int contLen, void *ahandle) { ...@@ -1222,6 +1222,7 @@ int taosSendMsgToPeerH(void *thandle, char *pCont, int contLen, void *ahandle) {
pServer = pConn->pServer; pServer = pConn->pServer;
pChann = pServer->channList + pConn->chann; pChann = pServer->channList + pConn->chann;
pHeader = (STaosHeader *)(pCont - sizeof(STaosHeader)); pHeader = (STaosHeader *)(pCont - sizeof(STaosHeader));
pHeader->destIp = pConn->peerIp;
msg = (char *)pHeader; msg = (char *)pHeader;
if ((pHeader->msgType & 1U) == 0 && pConn->localPort) pHeader->port = pConn->localPort; if ((pHeader->msgType & 1U) == 0 && pConn->localPort) pHeader->port = pConn->localPort;
......
src/system/detail/inc/mgmt.h
浏览文件 @ be2c6f5d
...@@ -222,6 +222,8 @@ typedef struct _connObj { ...@@ -222,6 +222,8 @@ typedef struct _connObj {
char superAuth : 1; // super user flag char superAuth : 1; // super user flag
char writeAuth : 1; // write flag char writeAuth : 1; // write flag
char killConnection : 1; // kill the connection flag char killConnection : 1; // kill the connection flag
uint8_t usePublicIp : 1; // if the connection request is publicIp
uint8_t reserved : 4;
uint32_t queryId; // query ID to be killed uint32_t queryId; // query ID to be killed
uint32_t streamId; // stream ID to be killed uint32_t streamId; // stream ID to be killed
uint32_t ip; // shell IP uint32_t ip; // shell IP
...@@ -343,7 +345,7 @@ void mgmtCleanUpVgroups(); ...@@ -343,7 +345,7 @@ void mgmtCleanUpVgroups();
int mgmtInitMeters(); int mgmtInitMeters();
STabObj *mgmtGetMeter(char *meterId); STabObj *mgmtGetMeter(char *meterId);
STabObj *mgmtGetMeterInfo(char *src, char *tags[]); 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 mgmtCreateMeter(SDbObj *pDb, SCreateTableMsg *pCreate);
int mgmtDropMeter(SDbObj *pDb, char *meterId, int ignore); int mgmtDropMeter(SDbObj *pDb, char *meterId, int ignore);
int mgmtAlterMeter(SDbObj *pDb, SAlterTableMsg *pAlter); int mgmtAlterMeter(SDbObj *pDb, SAlterTableMsg *pAlter);
......
src/system/detail/src/dnodeService.c
浏览文件 @ be2c6f5d
...@@ -55,12 +55,8 @@ int main(int argc, char *argv[]) { ...@@ -55,12 +55,8 @@ int main(int argc, char *argv[]) {
exit(EXIT_FAILURE); exit(EXIT_FAILURE);
} }
} else if (strcmp(argv[i], "-V") == 0) { } else if (strcmp(argv[i], "-V") == 0) {
#ifdef CLUSTER char *versionStr = tsIsCluster ? "enterprise" : "community";
printf("enterprise version: %s compatible_version: %s\n", version, compatible_version); printf("%s version: %s compatible_version: %s\n", versionStr, version, compatible_version);
#else
printf("community version: %s compatible_version: %s\n", version, compatible_version);
#endif
printf("gitinfo: %s\n", gitinfo); printf("gitinfo: %s\n", gitinfo);
printf("gitinfoI: %s\n", gitinfoOfInternal); printf("gitinfoI: %s\n", gitinfoOfInternal);
printf("buildinfo: %s\n", buildinfo); printf("buildinfo: %s\n", buildinfo);
......
src/system/detail/src/dnodeSystem.c
浏览文件 @ be2c6f5d
...@@ -136,7 +136,7 @@ int dnodeInitSystem() { ...@@ -136,7 +136,7 @@ int dnodeInitSystem() {
vnodeInitMgmtIp(); vnodeInitMgmtIp();
tsPrintGlobalConfig(); tsPrintGlobalConfig();
dPrint("Server IP address is:%s", tsInternalIp); dPrint("Server IP address is:%s", tsPrivateIp);
taosSetCoreDump(); taosSetCoreDump();
......
src/system/detail/src/mgmtMeter.c
浏览文件 @ be2c6f5d
...@@ -987,7 +987,7 @@ SSchema *mgmtGetMeterSchema(STabObj *pMeter) { ...@@ -987,7 +987,7 @@ SSchema *mgmtGetMeterSchema(STabObj *pMeter) {
/* /*
* serialize SVnodeSidList to byte array * 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, int32_t tagLen, int16_t numOfTags, int16_t *tagsId, int32_t maxNumOfMeters,
char *pMsg) { char *pMsg) {
if (pMeter->gid.vgId != *ovgId || ((*pList) != NULL && (*pList)->numOfSids >= maxNumOfMeters)) { if (pMeter->gid.vgId != *ovgId || ((*pList) != NULL && (*pList)->numOfSids >= maxNumOfMeters)) {
...@@ -1004,8 +1004,13 @@ static char *mgmtBuildMetricMetaMsg(STabObj *pMeter, int32_t *ovgId, SVnodeSidLi ...@@ -1004,8 +1004,13 @@ static char *mgmtBuildMetricMetaMsg(STabObj *pMeter, int32_t *ovgId, SVnodeSidLi
SVgObj *pVgroup = mgmtGetVgroup(pMeter->gid.vgId); SVgObj *pVgroup = mgmtGetVgroup(pMeter->gid.vgId);
for (int i = 0; i < TSDB_VNODES_SUPPORT; ++i) { for (int i = 0; i < TSDB_VNODES_SUPPORT; ++i) {
if (pConn->usePublicIp) {
(*pList)->vpeerDesc[i].ip = pVgroup->vnodeGid[i].publicIp; (*pList)->vpeerDesc[i].ip = pVgroup->vnodeGid[i].publicIp;
(*pList)->vpeerDesc[i].vnode = pVgroup->vnodeGid[i].vnode; (*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); pMsg += sizeof(SVnodeSidList);
...@@ -1105,10 +1110,10 @@ static SMetricMetaElemMsg *doConvertMetricMetaMsg(SMetricMetaMsg *pMetricMetaMsg ...@@ -1105,10 +1110,10 @@ static SMetricMetaElemMsg *doConvertMetricMetaMsg(SMetricMetaMsg *pMetricMetaMsg
return pElem; 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, char **pStart, int32_t *tagLen, int32_t rspMsgSize, int32_t maxTablePerVnode,
int32_t code) { 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) { if (*pStart == NULL) {
return 0; return 0;
} }
...@@ -1146,7 +1151,7 @@ static int32_t mgmtBuildMetricMetaRspMsg(void *thandle, SMetricMetaMsg *pMetricM ...@@ -1146,7 +1151,7 @@ static int32_t mgmtBuildMetricMetaRspMsg(void *thandle, SMetricMetaMsg *pMetricM
for (int32_t i = 0; i < pResult[j].num; ++i) { for (int32_t i = 0; i < pResult[j].num; ++i) {
STabObj *pMeter = pResult[j].pRes[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); maxTablePerVnode, pMsg);
} }
...@@ -1162,7 +1167,7 @@ static int32_t mgmtBuildMetricMetaRspMsg(void *thandle, SMetricMetaMsg *pMetricM ...@@ -1162,7 +1167,7 @@ static int32_t mgmtBuildMetricMetaRspMsg(void *thandle, SMetricMetaMsg *pMetricM
return msgLen; 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 * naive method: Do not limit the maximum number of meters in each
* vnode(subquery), split the result according to vnodes * vnode(subquery), split the result according to vnodes
...@@ -1236,8 +1241,7 @@ int mgmtRetrieveMetricMeta(void *thandle, char **pStart, SMetricMetaMsg *pMetric ...@@ -1236,8 +1241,7 @@ int mgmtRetrieveMetricMeta(void *thandle, char **pStart, SMetricMetaMsg *pMetric
msgLen = 512; msgLen = 512;
} }
msgLen = mgmtBuildMetricMetaRspMsg(thandle, pMetricMetaMsg, result, pStart, tagLen, msgLen, maxMetersPerVNodeForQuery, msgLen = mgmtBuildMetricMetaRspMsg(pConn, pMetricMetaMsg, result, pStart, tagLen, msgLen, maxMetersPerVNodeForQuery, ret);
ret);
for (int32_t i = 0; i < pMetricMetaMsg->numOfMeters; ++i) { for (int32_t i = 0; i < pMetricMetaMsg->numOfMeters; ++i) {
tQueryResultClean(&result[i]); tQueryResultClean(&result[i]);
......
src/system/detail/src/mgmtShell.c
浏览文件 @ be2c6f5d
...@@ -73,11 +73,8 @@ int mgmtInitShell() { ...@@ -73,11 +73,8 @@ int mgmtInitShell() {
if (numOfThreads < 1) numOfThreads = 1; if (numOfThreads < 1) numOfThreads = 1;
memset(&rpcInit, 0, sizeof(rpcInit)); memset(&rpcInit, 0, sizeof(rpcInit));
#ifdef CLUSTER
rpcInit.localIp = tsInternalIp; rpcInit.localIp = tsAnyIp ? "0.0.0.0" : tsPrivateIp;;
#else
rpcInit.localIp = "0.0.0.0";
#endif
rpcInit.localPort = tsMgmtShellPort; rpcInit.localPort = tsMgmtShellPort;
rpcInit.label = "MND-shell"; rpcInit.label = "MND-shell";
rpcInit.numOfThreads = numOfThreads; rpcInit.numOfThreads = numOfThreads;
...@@ -311,8 +308,13 @@ int mgmtProcessMeterMetaMsg(char *pMsg, int msgLen, SConnObj *pConn) { ...@@ -311,8 +308,13 @@ int mgmtProcessMeterMetaMsg(char *pMsg, int msgLen, SConnObj *pConn) {
goto _exit_code; goto _exit_code;
} }
for (int i = 0; i < TSDB_VNODES_SUPPORT; ++i) { for (int i = 0; i < TSDB_VNODES_SUPPORT; ++i) {
if (pConn->usePublicIp) {
pMeta->vpeerDesc[i].ip = pVgroup->vnodeGid[i].publicIp; pMeta->vpeerDesc[i].ip = pVgroup->vnodeGid[i].publicIp;
pMeta->vpeerDesc[i].vnode = htonl(pVgroup->vnodeGid[i].vnode); 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);
}
} }
} }
} }
...@@ -450,8 +452,13 @@ int mgmtProcessMultiMeterMetaMsg(char *pMsg, int msgLen, SConnObj *pConn) { ...@@ -450,8 +452,13 @@ int mgmtProcessMultiMeterMetaMsg(char *pMsg, int msgLen, SConnObj *pConn) {
} }
for (int i = 0; i < TSDB_VNODES_SUPPORT; ++i) { for (int i = 0; i < TSDB_VNODES_SUPPORT; ++i) {
if (pConn->usePublicIp) {
pMeta->meta.vpeerDesc[i].ip = pVgroup->vnodeGid[i].publicIp; pMeta->meta.vpeerDesc[i].ip = pVgroup->vnodeGid[i].publicIp;
pMeta->meta.vpeerDesc[i].vnode = htonl(pVgroup->vnodeGid[i].vnode); 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);
}
} }
} }
} }
...@@ -523,7 +530,7 @@ int mgmtProcessMetricMetaMsg(char *pMsg, int msgLen, SConnObj *pConn) { ...@@ -523,7 +530,7 @@ int mgmtProcessMetricMetaMsg(char *pMsg, int msgLen, SConnObj *pConn) {
msgLen = pMsg - pStart; msgLen = pMsg - pStart;
} else { } else {
msgLen = mgmtRetrieveMetricMeta(pConn->thandle, &pStart, pMetricMetaMsg); msgLen = mgmtRetrieveMetricMeta(pConn, &pStart, pMetricMetaMsg);
if (msgLen <= 0) { if (msgLen <= 0) {
taosSendSimpleRsp(pConn->thandle, TSDB_MSG_TYPE_METRIC_META_RSP, TSDB_CODE_SERV_OUT_OF_MEMORY); taosSendSimpleRsp(pConn->thandle, TSDB_MSG_TYPE_METRIC_META_RSP, TSDB_CODE_SERV_OUT_OF_MEMORY);
return 0; return 0;
...@@ -1103,10 +1110,17 @@ int mgmtProcessHeartBeatMsg(char *cont, int contLen, SConnObj *pConn) { ...@@ -1103,10 +1110,17 @@ int mgmtProcessHeartBeatMsg(char *cont, int contLen, SConnObj *pConn) {
pHBRsp->killConnection = pConn->killConnection; pHBRsp->killConnection = pConn->killConnection;
#ifdef CLUSTER #ifdef CLUSTER
if (pConn->usePublicIp) {
int size = pSdbPublicIpList->numOfIps * 4; int size = pSdbPublicIpList->numOfIps * 4;
pHBRsp->ipList.numOfIps = pSdbPublicIpList->numOfIps; pHBRsp->ipList.numOfIps = pSdbPublicIpList->numOfIps;
memcpy(pHBRsp->ipList.ip, pSdbPublicIpList->ip, size); memcpy(pHBRsp->ipList.ip, pSdbPublicIpList->ip, size);
pMsg += sizeof(SHeartBeatRsp) + 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 #else
pMsg += sizeof(SHeartBeatRsp); pMsg += sizeof(SHeartBeatRsp);
#endif #endif
...@@ -1183,6 +1197,12 @@ int mgmtProcessConnectMsg(char *pMsg, int msgLen, SConnObj *pConn) { ...@@ -1183,6 +1197,12 @@ int mgmtProcessConnectMsg(char *pMsg, int msgLen, SConnObj *pConn) {
pAcct = mgmtGetAcct(pUser->acct); pAcct = mgmtGetAcct(pUser->acct);
code = taosCheckVersion(pConnectMsg->clientVersion, version, 3);
if (code != 0) {
mError("invalid client version:%s", pConnectMsg->clientVersion);
goto _rsp;
}
if (pConnectMsg->db[0]) { if (pConnectMsg->db[0]) {
sprintf(dbName, "%x%s%s", pAcct->acctId, TS_PATH_DELIMITER, pConnectMsg->db); sprintf(dbName, "%x%s%s", pAcct->acctId, TS_PATH_DELIMITER, pConnectMsg->db);
pDb = mgmtGetDb(dbName); pDb = mgmtGetDb(dbName);
...@@ -1223,7 +1243,11 @@ _rsp: ...@@ -1223,7 +1243,11 @@ _rsp:
#ifdef CLUSTER #ifdef CLUSTER
int size = pSdbPublicIpList->numOfIps * 4 + sizeof(SIpList); int size = pSdbPublicIpList->numOfIps * 4 + sizeof(SIpList);
if (pConn->usePublicIp) {
memcpy(pMsg, pSdbPublicIpList, size); memcpy(pMsg, pSdbPublicIpList, size);
} else {
memcpy(pMsg, pSdbIpList, size);
}
pMsg += size; pMsg += size;
#endif #endif
...@@ -1273,6 +1297,9 @@ void *mgmtProcessMsgFromShell(char *msg, void *ahandle, void *thandle) { ...@@ -1273,6 +1297,9 @@ void *mgmtProcessMsgFromShell(char *msg, void *ahandle, void *thandle) {
pConn = connList + pMsg->destId; pConn = connList + pMsg->destId;
pConn->thandle = thandle; pConn->thandle = thandle;
strcpy(pConn->user, pMsg->meterId); strcpy(pConn->user, pMsg->meterId);
pConn->usePublicIp = (pMsg->destIp == tsPublicIpInt ? 1 : 0);
mTrace("pConn:%p is rebuild, destIp:%s publicIp:%s usePublicIp:%u",
pConn, taosIpStr(pMsg->destIp), taosIpStr(tsPublicIpInt), pConn->usePublicIp);
} }
if (pMsg->msgType == TSDB_MSG_TYPE_CONNECT) { if (pMsg->msgType == TSDB_MSG_TYPE_CONNECT) {
......
src/system/detail/src/mgmtVgroup.c
浏览文件 @ be2c6f5d
...@@ -102,13 +102,17 @@ int mgmtInitVgroups() { ...@@ -102,13 +102,17 @@ int mgmtInitVgroups() {
} }
taosIdPoolReinit(pVgroup->idPool); taosIdPoolReinit(pVgroup->idPool);
#ifdef CLUSTER
if (tsIsCluster) {
/*
* Upgrade from open source version to cluster version for the first time
*/
if (pVgroup->vnodeGid[0].publicIp == 0) { if (pVgroup->vnodeGid[0].publicIp == 0) {
pVgroup->vnodeGid[0].publicIp = inet_addr(tsPublicIp); pVgroup->vnodeGid[0].publicIp = inet_addr(tsPublicIp);
pVgroup->vnodeGid[0].ip = inet_addr(tsPrivateIp); pVgroup->vnodeGid[0].ip = inet_addr(tsPrivateIp);
sdbUpdateRow(vgSdb, pVgroup, tsVgUpdateSize, 1); sdbUpdateRow(vgSdb, pVgroup, tsVgUpdateSize, 1);
} }
#endif }
mgmtSetDnodeVgid(pVgroup->vnodeGid, pVgroup->numOfVnodes, pVgroup->vgId); mgmtSetDnodeVgid(pVgroup->vnodeGid, pVgroup->numOfVnodes, pVgroup->vgId);
} }
......
src/system/detail/src/vnodeShell.c
浏览文件 @ be2c6f5d
...@@ -142,11 +142,9 @@ int vnodeInitShell() { ...@@ -142,11 +142,9 @@ int vnodeInitShell() {
if (numOfThreads < 1) numOfThreads = 1; if (numOfThreads < 1) numOfThreads = 1;
memset(&rpcInit, 0, sizeof(rpcInit)); memset(&rpcInit, 0, sizeof(rpcInit));
#ifdef CLUSTER
rpcInit.localIp = tsInternalIp; rpcInit.localIp = tsAnyIp ? "0.0.0.0" : tsPrivateIp;
#else
rpcInit.localIp = "0.0.0.0";
#endif
rpcInit.localPort = tsVnodeShellPort; rpcInit.localPort = tsVnodeShellPort;
rpcInit.label = "DND-shell"; rpcInit.label = "DND-shell";
rpcInit.numOfThreads = numOfThreads; rpcInit.numOfThreads = numOfThreads;
......
src/system/lite/src/mgmtDnode.spec.c
浏览文件 @ be2c6f5d
...@@ -27,7 +27,7 @@ int mgmtUpdateDnode(SDnodeObj *pDnode) { return 0; } ...@@ -27,7 +27,7 @@ int mgmtUpdateDnode(SDnodeObj *pDnode) { return 0; }
void mgmtCleanUpDnodes() {} void mgmtCleanUpDnodes() {}
int mgmtInitDnodes() { int mgmtInitDnodes() {
dnodeObj.privateIp = inet_addr(tsInternalIp);; dnodeObj.privateIp = inet_addr(tsPrivateIp);;
dnodeObj.createdTime = (int64_t)tsRebootTime * 1000; dnodeObj.createdTime = (int64_t)tsRebootTime * 1000;
dnodeObj.lastReboot = tsRebootTime; dnodeObj.lastReboot = tsRebootTime;
dnodeObj.numOfCores = (uint16_t)tsNumOfCores; dnodeObj.numOfCores = (uint16_t)tsNumOfCores;
......
src/util/src/tglobalcfg.c
浏览文件 @ be2c6f5d
...@@ -75,7 +75,6 @@ int tsMetricMetaKeepTimer = 600; // second ...@@ -75,7 +75,6 @@ int tsMetricMetaKeepTimer = 600; // second
float tsNumOfThreadsPerCore = 1.0; float tsNumOfThreadsPerCore = 1.0;
float tsRatioOfQueryThreads = 0.5; float tsRatioOfQueryThreads = 0.5;
char tsPublicIp[TSDB_IPv4ADDR_LEN] = {0}; char tsPublicIp[TSDB_IPv4ADDR_LEN] = {0};
char tsInternalIp[TSDB_IPv4ADDR_LEN] = {0};
char tsPrivateIp[TSDB_IPv4ADDR_LEN] = {0}; char tsPrivateIp[TSDB_IPv4ADDR_LEN] = {0};
char tsServerIpStr[TSDB_IPv4ADDR_LEN] = "127.0.0.1"; char tsServerIpStr[TSDB_IPv4ADDR_LEN] = "127.0.0.1";
short tsNumOfVnodesPerCore = 8; short tsNumOfVnodesPerCore = 8;
...@@ -165,6 +164,14 @@ int tsAdminRowLimit = 10240; ...@@ -165,6 +164,14 @@ int tsAdminRowLimit = 10240;
int tsTscEnableRecordSql = 0; int tsTscEnableRecordSql = 0;
int tsEnableCoreFile = 0; int tsEnableCoreFile = 0;
int tsAnyIp = 1;
uint32_t tsPublicIpInt = 0;
#ifdef CLUSTER
int tsIsCluster = 1;
#else
int tsIsCluster = 0;
#endif
int tsRpcTimer = 300; int tsRpcTimer = 300;
int tsRpcMaxTime = 600; // seconds; int tsRpcMaxTime = 600; // seconds;
...@@ -449,9 +456,6 @@ static void doInitGlobalConfig() { ...@@ -449,9 +456,6 @@ static void doInitGlobalConfig() {
tsInitConfigOption(cfg++, "privateIp", tsPrivateIp, TSDB_CFG_VTYPE_IPSTR, tsInitConfigOption(cfg++, "privateIp", tsPrivateIp, TSDB_CFG_VTYPE_IPSTR,
TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_CLUSTER, TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_CLUSTER,
0, 0, TSDB_IPv4ADDR_LEN, TSDB_CFG_UTYPE_NONE); 0, 0, TSDB_IPv4ADDR_LEN, TSDB_CFG_UTYPE_NONE);
tsInitConfigOption(cfg++, "internalIp", tsInternalIp, TSDB_CFG_VTYPE_IPSTR,
TSDB_CFG_CTYPE_B_CONFIG,
0, 0, TSDB_IPv4ADDR_LEN, TSDB_CFG_UTYPE_NONE);
tsInitConfigOption(cfg++, "localIp", tsLocalIp, TSDB_CFG_VTYPE_IPSTR, tsInitConfigOption(cfg++, "localIp", tsLocalIp, TSDB_CFG_VTYPE_IPSTR,
TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_CLIENT, TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_CLIENT,
0, 0, TSDB_IPv4ADDR_LEN, TSDB_CFG_UTYPE_NONE); 0, 0, TSDB_IPv4ADDR_LEN, TSDB_CFG_UTYPE_NONE);
...@@ -714,7 +718,7 @@ static void doInitGlobalConfig() { ...@@ -714,7 +718,7 @@ static void doInitGlobalConfig() {
1, 100000, 0, TSDB_CFG_UTYPE_NONE); 1, 100000, 0, TSDB_CFG_UTYPE_NONE);
tsInitConfigOption(cfg++, "httpEnableRecordSql", &tsHttpEnableRecordSql, TSDB_CFG_VTYPE_INT, tsInitConfigOption(cfg++, "httpEnableRecordSql", &tsHttpEnableRecordSql, TSDB_CFG_VTYPE_INT,
TSDB_CFG_CTYPE_B_CONFIG, TSDB_CFG_CTYPE_B_CONFIG,
1, 100000, 0, TSDB_CFG_UTYPE_NONE); 0, 1, 0, TSDB_CFG_UTYPE_NONE);
tsInitConfigOption(cfg++, "telegrafUseFieldNum", &tsTelegrafUseFieldNum, TSDB_CFG_VTYPE_INT, tsInitConfigOption(cfg++, "telegrafUseFieldNum", &tsTelegrafUseFieldNum, TSDB_CFG_VTYPE_INT,
TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW, TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_SHOW,
0, 1, 1, TSDB_CFG_UTYPE_NONE); 0, 1, 1, TSDB_CFG_UTYPE_NONE);
...@@ -778,11 +782,17 @@ static void doInitGlobalConfig() { ...@@ -778,11 +782,17 @@ static void doInitGlobalConfig() {
tsInitConfigOption(cfg++, "tscEnableRecordSql", &tsTscEnableRecordSql, TSDB_CFG_VTYPE_INT, tsInitConfigOption(cfg++, "tscEnableRecordSql", &tsTscEnableRecordSql, TSDB_CFG_VTYPE_INT,
TSDB_CFG_CTYPE_B_CONFIG, TSDB_CFG_CTYPE_B_CONFIG,
1, 100000, 0, TSDB_CFG_UTYPE_NONE); 0, 1, 0, TSDB_CFG_UTYPE_NONE);
tsInitConfigOption(cfg++, "enableCoreFile", &tsEnableCoreFile, TSDB_CFG_VTYPE_INT, tsInitConfigOption(cfg++, "enableCoreFile", &tsEnableCoreFile, TSDB_CFG_VTYPE_INT,
TSDB_CFG_CTYPE_B_CONFIG, TSDB_CFG_CTYPE_B_CONFIG,
1, 100000, 0, TSDB_CFG_UTYPE_NONE); 0, 1, 0, TSDB_CFG_UTYPE_NONE);
#ifdef CLUSTER
tsInitConfigOption(cfg++, "anyIp", &tsAnyIp, TSDB_CFG_VTYPE_INT,
TSDB_CFG_CTYPE_B_CONFIG,
0, 1, 0, TSDB_CFG_UTYPE_NONE);
#endif
// version info // version info
tsInitConfigOption(cfg++, "gitinfo", gitinfo, TSDB_CFG_VTYPE_STRING, tsInitConfigOption(cfg++, "gitinfo", gitinfo, TSDB_CFG_VTYPE_STRING,
...@@ -911,10 +921,7 @@ bool tsReadGlobalConfig() { ...@@ -911,10 +921,7 @@ bool tsReadGlobalConfig() {
if (tsPublicIp[0] == 0) { if (tsPublicIp[0] == 0) {
strcpy(tsPublicIp, tsPrivateIp); strcpy(tsPublicIp, tsPrivateIp);
} }
tsPublicIpInt = inet_addr(tsPublicIp);
if (tsInternalIp[0] == 0) {
strcpy(tsInternalIp, tsPrivateIp);
}
if (tsLocalIp[0] == 0) { if (tsLocalIp[0] == 0) {
strcpy(tsLocalIp, tsPrivateIp); strcpy(tsLocalIp, tsPrivateIp);
......
src/util/src/tutil.c
浏览文件 @ be2c6f5d
...@@ -24,6 +24,8 @@ ...@@ -24,6 +24,8 @@
#include "ttime.h" #include "ttime.h"
#include "ttypes.h" #include "ttypes.h"
#include "tutil.h" #include "tutil.h"
#include "tlog.h"
#include "taoserror.h"
int32_t strdequote(char *z) { int32_t strdequote(char *z) {
if (z == NULL) { if (z == NULL) {
...@@ -451,10 +453,8 @@ bool taosValidateEncodec(const char *encodec) { ...@@ -451,10 +453,8 @@ bool taosValidateEncodec(const char *encodec) {
return false; return false;
} }
iconv_close(cd); iconv_close(cd);
return true;
#else
return true;
#endif #endif
return true;
} }
bool taosGetVersionNumber(char *versionStr, int *versionNubmer) { bool taosGetVersionNumber(char *versionStr, int *versionNubmer) {
...@@ -486,6 +486,36 @@ bool taosGetVersionNumber(char *versionStr, int *versionNubmer) { ...@@ -486,6 +486,36 @@ bool taosGetVersionNumber(char *versionStr, int *versionNubmer) {
return true; 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) { char *taosIpStr(uint32_t ipInt) {
static char ipStrArray[3][30]; static char ipStrArray[3][30];
static int ipStrIndex = 0; static int ipStrIndex = 0;
......
Markdown is supported
0% 或 .
You are about to add 0 people to the discussion. Proceed with caution.
先完成此消息的编辑！
想要评论请 注册 或