Merge pull request #1249 from taosdata/refact/rpc

add destIp support for NAT

src/inc/mnode.h

@@ -254,6 +254,8 @@ typedef struct _user_obj {
   char              reserved[16];
   char              updateEnd[1];
   struct _user_obj *prev, *next;
+  int8_t            writeAuth;
+  int8_t            superAuth;
 } SUserObj;
 
 typedef struct {

src/inc/taosmsg.h

@@ -187,6 +187,13 @@ typedef enum {
 
 extern char *taosMsg[];
 
+#define TSDB_MSG_DEF_MAX_MPEERS        5
+#define TSDB_MSG_DEF_VERSION_LEN       64
+#define TSDB_MSG_DEF_DB_LEN            128
+#define TSDB_MSG_DEF_USER_LEN          128
+#define TSDB_MSG_DEF_TABLE_LEN         128
+#define TSDB_MSG_DEF_ACCT_LEN          128
+
 #pragma pack(push, 1)
 
 typedef struct {
@@ -325,9 +332,23 @@ typedef struct {
 } SAlterTableMsg;
 
 typedef struct {
-  char clientVersion[TSDB_VERSION_LEN];
-  char db[TSDB_TABLE_ID_LEN];
-} SConnectMsg;
+  char clientVersion[TSDB_MSG_DEF_VERSION_LEN];
+  char msgVersion[TSDB_MSG_DEF_VERSION_LEN];
+  char db[TSDB_MSG_DEF_DB_LEN];
+} SCMConnectMsg;
+
+typedef struct {
+  char     acctId[TSDB_MSG_DEF_ACCT_LEN];
+  char     serverVersion[TSDB_MSG_DEF_VERSION_LEN];
+  int8_t   writeAuth;
+  int8_t   superAuth;
+  int8_t   usePublicIp;
+  int16_t  index;
+  int16_t  numOfIps;
+  uint16_t port;
+  uint32_t ip[TSDB_MSG_DEF_MAX_MPEERS];
+} SCMConnectRsp;
+
 
 typedef struct {
   int32_t maxUsers;
@@ -360,13 +381,6 @@ typedef struct {
   char db[TSDB_TABLE_ID_LEN];
 } SMgmtHead;
 
-typedef struct {
-  char acctId[TSDB_ACCT_LEN];
-  char version[TSDB_VERSION_LEN];
-  char writeAuth;
-  char superAuth;
-} SConnectRsp;
-
 typedef struct {
   short    vnode;
   int32_t  sid;

src/inc/trpc.h

@@ -46,8 +46,9 @@ typedef struct {
 } SRpcIpSet;
 
 typedef struct {
-  uint32_t  sourceIp;
-  uint16_t  sourcePort;
+  uint32_t  clientIp;
+  uint16_t  clientPort;
+  uint32_t  serverIp;
   char     *user;
 } SRpcConnInfo;
 

src/mnode/inc/mgmtShell.h

@@ -24,7 +24,7 @@ extern "C" {
 #include <stdbool.h>
 #include "mnode.h"
 
-int  mgmtInitShell();
+int32_t  mgmtInitShell();
 void mgmtCleanUpShell();
 
 extern int32_t (*mgmtCheckRedirectMsg)(SConnObj *pConn, int32_t msgType);

src/rpc/inc/rpcHead.h

@@ -32,6 +32,7 @@ typedef struct {
   uint32_t uid;       // for unique ID inside a client
   uint32_t sourceId;  // source ID, an index for connection list  
   uint32_t destId;    // destination ID, an index for connection list
+  uint32_t destIp;    // destination IP address, for NAT scenario
   char     user[TSDB_UNI_LEN];
   uint16_t port;      // for UDP only, port may be changed
   char     empty[1];  // reserved

src/rpc/src/rpcMain.c

@@ -94,6 +94,7 @@ typedef struct _RpcConn {
   uint16_t  localPort;      // for UDP only
   uint32_t  peerUid;        // peer UID
   uint32_t  peerIp;         // peer IP
+  uint32_t  destIp;         // server destination IP to handle NAT 
   uint16_t  peerPort;       // peer port
   char      peerIpstr[TSDB_IPv4ADDR_LEN];  // peer IP string
   uint16_t  tranId;         // outgoing transcation ID, for build message
@@ -389,8 +390,9 @@ void rpcGetConnInfo(void *thandle, SRpcConnInfo *pInfo) {
   SRpcConn  *pConn = (SRpcConn *)thandle;
   SRpcInfo  *pRpc = pConn->pRpc;
 
-  pInfo->sourceIp = pConn->peerIp;
-  pInfo->sourcePort = pConn->peerPort;
+  pInfo->clientIp = pConn->peerIp;
+  pInfo->clientPort = pConn->peerPort;
+  pInfo->serverIp = pConn->destIp;
   strcpy(pInfo->user, pConn->user);
 }
 
@@ -546,6 +548,7 @@ SRpcConn *rpcSetConnToServer(SRpcInfo *pRpc, SRpcIpSet ipSet) {
     char ipstr[20] = {0};
     tinet_ntoa(ipstr, ipSet.ip[ipSet.index]);
     pConn = rpcOpenConn(pRpc, ipstr, ipSet.port);
+    pConn->destIp = ipSet.ip[ipSet.index];
   } 
 
   return pConn;
@@ -772,11 +775,13 @@ static void *rpcProcessMsgFromPeer(void *msg, int msgLen, uint32_t ip, uint16_t
 
 static void rpcProcessIncomingMsg(SRpcConn *pConn, SRpcHead *pHead) {
   SRpcInfo *pRpc = pConn->pRpc;
+
   pHead = rpcDecompressRpcMsg(pHead);
-  int       contLen = rpcContLenFromMsg(pHead->msgLen);
-  uint8_t  *pCont = pHead->content;
+  int     contLen = rpcContLenFromMsg(pHead->msgLen);
+  uint8_t *pCont = pHead->content;
    
   if ( rpcIsReq(pHead->msgType) ) {
+    pConn->destIp = pHead->destIp;
     taosTmrReset(rpcProcessProgressTimer, tsRpcTimer/2, pConn, pRpc->tmrCtrl, &pConn->pTimer);
     (*(pRpc->cfp))(pHead->msgType, pCont, contLen, pConn, 0);
   } else {
@@ -886,6 +891,7 @@ static void rpcSendReqToServer(SRpcInfo *pRpc, SRpcReqContext *pContext) {
   pHead->tranId = pConn->tranId;
   pHead->sourceId = pConn->ownId;
   pHead->destId = pConn->peerId;
+  pHead->destIp = pConn->destIp;
   pHead->port = 0;
   pHead->uid = (uint32_t)((int64_t)pConn + (int64_t)getpid());
   memcpy(pHead->user, pConn->user, tListLen(pHead->user));