[TD-464] support kill connection

src/client/src/tscSQLParser.c

@@ -93,7 +93,7 @@ static int32_t validateArithmeticSQLExpr(tSQLExpr* pExpr, SQueryInfo* pQueryInfo
 static int32_t validateDNodeConfig(tDCLSQL* pOptions);
 static int32_t validateLocalConfig(tDCLSQL* pOptions);
 static int32_t validateColumnName(char* name);
-static int32_t setKillInfo(SSqlObj* pSql, struct SSqlInfo* pInfo);
+static int32_t setKillInfo(SSqlObj* pSql, struct SSqlInfo* pInfo, int32_t killType);
 
 static bool validateOneTags(SSqlCmd* pCmd, TAOS_FIELD* pTagField);
 static bool hasTimestampForPointInterpQuery(SQueryInfo* pQueryInfo);
@@ -532,7 +532,7 @@ int32_t tscToSQLCmd(SSqlObj* pSql, struct SSqlInfo* pInfo) {
     case TSDB_SQL_KILL_QUERY:
     case TSDB_SQL_KILL_STREAM:
     case TSDB_SQL_KILL_CONNECTION: {
-      if ((code = setKillInfo(pSql, pInfo)) != TSDB_CODE_SUCCESS) {
+      if ((code = setKillInfo(pSql, pInfo, pInfo->type)) != TSDB_CODE_SUCCESS) {
         return code;
       }
 
@@ -2231,31 +2231,46 @@ int32_t setShowInfo(SSqlObj* pSql, struct SSqlInfo* pInfo) {
   return TSDB_CODE_SUCCESS;
 }
 
-int32_t setKillInfo(SSqlObj* pSql, struct SSqlInfo* pInfo) {
-  const char* msg1 = "invalid ip address";
-  const char* msg2 = "invalid port";
+int32_t setKillInfo(SSqlObj* pSql, struct SSqlInfo* pInfo, int32_t killType) {
+  const char* msg1 = "invalid connection ID";
+  const char* msg2 = "invalid query ID";
+  const char* msg3 = "invalid stream ID";
 
   SSqlCmd* pCmd = &pSql->cmd;
   pCmd->command = pInfo->type;
 
-  SSQLToken* ip = &(pInfo->pDCLInfo->ip);
-  if (ip->n > TSDB_KILL_MSG_LEN) {
+  SSQLToken* idStr = &(pInfo->pDCLInfo->ip);
+  if (idStr->n > TSDB_KILL_MSG_LEN) {
     return TSDB_CODE_INVALID_SQL;
   }
 
-  strncpy(pCmd->payload, ip->z, ip->n);
-
+  strncpy(pCmd->payload, idStr->z, idStr->n);
   const char delim = ':';
+  char* connIdStr = strtok(idStr->z, &delim);
+  char* queryIdStr = strtok(NULL, &delim);
+
+  int32_t connId = (int32_t)strtol(connIdStr, NULL, 10);
+  if (connId <= 0) {
+    memset(pCmd->payload, 0, strlen(pCmd->payload));
+    return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg1);
+  }
 
-  char* ipStr = strtok(ip->z, &delim);
-  char* portStr = strtok(NULL, &delim);
+  if (killType == TSDB_SQL_KILL_CONNECTION) {
+    strncpy(pCmd->payload, idStr->z, idStr->n);
+    return TSDB_CODE_SUCCESS;
+  }
 
-  uint16_t port = (uint16_t)strtol(portStr, NULL, 10);
-  if (port <= 0 || port > 65535) {
+  int32_t queryId = (int32_t)strtol(queryIdStr, NULL, 10);
+  if (queryId <= 0) {
     memset(pCmd->payload, 0, strlen(pCmd->payload));
+    if (killType == TSDB_SQL_KILL_QUERY) {
       return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg2);
+    } else {
+      return invalidSqlErrMsg(tscGetErrorMsgPayload(pCmd), msg3);
+    }
   }
 
+  strncpy(pCmd->payload, idStr->z, idStr->n);
   return TSDB_CODE_SUCCESS;
 }
 

src/inc/ttokendef.h

@@ -148,8 +148,8 @@
 #define TK_SET                            130
 #define TK_KILL                           131
 #define TK_CONNECTION                     132
-#define TK_COLON                          133
-#define TK_STREAM                         134
+#define TK_STREAM                         133
+#define TK_COLON                          134
 #define TK_ABORT                          135
 #define TK_AFTER                          136
 #define TK_ATTACH                         137

src/mnode/inc/mnodeProfile.h

@@ -21,11 +21,22 @@ extern "C" {
 #endif
 #include "mnodeDef.h"
 
+typedef struct {
+  char     user[TSDB_USER_LEN + 1];
+  int8_t   killed;
+  uint16_t port;
+  uint32_t ip;
+  uint32_t connId;
+  uint64_t stime;
+  uint64_t lastAccess;
+} SConnObj;
+
 int32_t mnodeInitProfile();
 void    mnodeCleanupProfile();
 
-uint32_t mnodeCreateConn(char *user, uint32_t ip, uint16_t port);
-bool     mnodeCheckConn(uint32_t connId, char *user, uint32_t ip, uint16_t port);
+SConnObj *mnodeCreateConn(char *user, uint32_t ip, uint16_t port);
+SConnObj *mnodeAccquireConn(uint32_t connId, char *user, uint32_t ip, uint16_t port);
+void      mnodeReleaseConn(SConnObj *pConn);
 
 #ifdef __cplusplus
 }

src/mnode/src/mnodeProfile.c

@@ -39,15 +39,6 @@
 #define CONN_KEEP_TIME (tsShellActivityTimer * 3)
 #define CONN_CHECK_TIME (tsShellActivityTimer * 2)
 
-typedef struct {
-  char     user[TSDB_USER_LEN + 1];
-  int8_t   killed;
-  uint16_t port;
-  uint32_t ip;
-  uint32_t connId;
-  uint64_t stime;
-} SConnObj;
-
 extern void *tsMnodeTmr;
 static SCacheObj *tsMnodeConnCache = NULL;
 static uint32_t tsConnIndex = 0;
@@ -91,13 +82,13 @@ void mnodeCleanupProfile() {
   }
 }
 
-uint32_t mnodeCreateConn(char *user, uint32_t ip, uint16_t port) {
+SConnObj *mnodeCreateConn(char *user, uint32_t ip, uint16_t port) {
   int32_t connSize = taosHashGetSize(tsMnodeConnCache->pHashTable);
   if (connSize > tsMaxShellConns) {
     mError("failed to create conn for user:%s ip:%s:%u, conns:%d larger than maxShellConns:%d, ", user, taosIpStr(ip),
            port, connSize, tsMaxShellConns);
     terrno = TSDB_CODE_TOO_MANY_SHELL_CONNS;
-    return 0;
+    return NULL;
   }
 
   uint32_t connId = atomic_add_fetch_32(&tsConnIndex, 1);
@@ -109,18 +100,22 @@ uint32_t mnodeCreateConn(char *user, uint32_t ip, uint16_t port) {
     .connId = connId,
     .stime  = taosGetTimestampMs()
   };
+  strcpy(connObj.user, user);
   
   char key[10];
   sprintf(key, "%u", connId);  
-  strcpy(connObj.user, user);
-  void *pConn = taosCachePut(tsMnodeConnCache, key, &connObj, sizeof(connObj), CONN_KEEP_TIME);
-  taosCacheRelease(tsMnodeConnCache, &pConn, false);
+  SConnObj *pConn = taosCachePut(tsMnodeConnCache, key, &connObj, sizeof(connObj), CONN_KEEP_TIME);
   
   mTrace("connId:%d, is created, user:%s ip:%s:%u", connId, user, taosIpStr(ip), port);
-  return connId;
+  return pConn;
+}
+
+void mnodeReleaseConn(SConnObj *pConn) {
+  if(pConn == NULL) return;
+  taosCacheRelease(tsMnodeConnCache, (void**)&pConn, false);
 }
 
-bool mnodeCheckConn(uint32_t connId, char *user, uint32_t ip, uint16_t port) {
+SConnObj *mnodeAccquireConn(uint32_t connId, char *user, uint32_t ip, uint16_t port) {
   char key[10];
   sprintf(key, "%u", connId);
   uint64_t expireTime = CONN_KEEP_TIME * 1000 + (uint64_t)taosGetTimestampMs();
@@ -128,19 +123,19 @@ bool mnodeCheckConn(uint32_t connId, char *user, uint32_t ip, uint16_t port) {
   SConnObj *pConn = taosCacheUpdateExpireTimeByName(tsMnodeConnCache, key, expireTime);
   if (pConn == NULL) {
     mError("connId:%d, is already destroyed, user:%s ip:%s:%u", connId, user, taosIpStr(ip), port);
-    return false;
+    return NULL;
   }
 
-  if (pConn->ip != ip || pConn->port != port /* || strcmp(pConn->user, user) != 0 */ ) {
+  if (pConn->ip != ip || pConn->port != port /* || strcmp(pConn->user, user) != 0 */) {
     mError("connId:%d, incoming conn user:%s ip:%s:%u, not match exist conn user:%s ip:%s:%u", connId, user,
            taosIpStr(ip), port, pConn->user, taosIpStr(pConn->ip), pConn->port);
-    taosCacheRelease(tsMnodeConnCache, (void**)&pConn, false);       
-    return false;
+    taosCacheRelease(tsMnodeConnCache, (void **)&pConn, false);
+    return NULL;
   }
 
-  //mTrace("connId:%d, is incoming, user:%s ip:%s:%u", connId, pConn->user, taosIpStr(pConn->ip), pConn->port);
-  taosCacheRelease(tsMnodeConnCache, (void**)&pConn, false);
-  return true;
+  // mTrace("connId:%d, is incoming, user:%s ip:%s:%u", connId, pConn->user, taosIpStr(pConn->ip), pConn->port);
+  pConn->lastAccess = expireTime;
+  return pConn;
 }
 
 static void mnodeFreeConn(void *data) {
@@ -553,6 +548,12 @@ int32_t mnodeGetConnsMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn) {
   pSchema[cols].bytes = htons(pShow->bytes[cols]);
   cols++;
 
+  pShow->bytes[cols] = 8;
+  pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP;
+  strcpy(pSchema[cols].name, "last access");
+  pSchema[cols].bytes = htons(pShow->bytes[cols]);
+  cols++;
+
   pMeta->numOfColumns = htons(cols);
   pShow->numOfColumns = cols;
 
@@ -597,6 +598,10 @@ int32_t mnodeRetrieveConns(SShowObj *pShow, char *data, int32_t rows, void *pCon
     *(int64_t *)pWrite = pConnObj->stime;
     cols++;
 
+    pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
+    *(int64_t *)pWrite = pConnObj->lastAccess;
+    cols++;
+
     numOfRows++;
   }
 
@@ -675,7 +680,7 @@ int32_t mnodeProcessKillConnectionMsg(SMnodeMsg *pMsg) {
     mError("connId:%s, failed to kill, conn not exist", pKill->queryId);
     return TSDB_CODE_INVALID_CONNECTION;
   } else {
-    mError("connId:%s, is killed by user:%s", pKill->queryId, pUser->user);
+    mPrint("connId:%s, is killed by user:%s", pKill->queryId, pUser->user);
     pConn->killed = 1;
     taosCacheRelease(tsMnodeConnCache, (void**)&pConn, false);
     return TSDB_CODE_SUCCESS;

src/mnode/src/mnodeShow.c

@@ -232,18 +232,22 @@ static int32_t mnodeProcessHeartBeatMsg(SMnodeMsg *pMsg) {
   rpcGetConnInfo(pMsg->rpcMsg.handle, &connInfo);
     
   int32_t connId = htonl(pHBMsg->connId);
-  if (!mnodeCheckConn(connId, connInfo.user, connInfo.clientIp, connInfo.clientPort)) {
-    connId = mnodeCreateConn(connInfo.user, connInfo.clientIp, connInfo.clientPort);
-    if (connId == 0) {
-#if 0 
+  SConnObj *pConn = mnodeAccquireConn(connId, connInfo.user, connInfo.clientIp, connInfo.clientPort);
+  if (pConn == NULL) {
+    pConn = mnodeCreateConn(connInfo.user, connInfo.clientIp, connInfo.clientPort);
+  }
+
+  if (pConn == NULL) {
     // do not close existing links, otherwise
-      mError("failed to create connId, close connect");
+    // mError("failed to create connId, close connect");
+    // pHBRsp->killConnection = 1;
+  } else {
+    pHBRsp->connId = htonl(pConn->connId);
+    if (pConn->killed != 0) {
       pHBRsp->killConnection = 1;
-#endif      
     }
   }
 
-  pHBRsp->connId = htonl(connId);
   pHBRsp->onlineDnodes = htonl(mnodeGetOnlinDnodesNum());
   pHBRsp->totalDnodes = htonl(mnodeGetDnodesNum());
   mnodeGetMnodeIpSetForShell(&pHBRsp->ipList);
@@ -259,6 +263,7 @@ static int32_t mnodeProcessHeartBeatMsg(SMnodeMsg *pMsg) {
   pMsg->rpcRsp.rsp = pHBRsp;
   pMsg->rpcRsp.len = sizeof(SCMHeartBeatRsp);
   
+  mnodeReleaseConn(pConn);
   return TSDB_CODE_SUCCESS;
 }
 
@@ -298,14 +303,18 @@ static int32_t mnodeProcessConnectMsg(SMnodeMsg *pMsg) {
     goto connect_over;
   }
 
-  int32_t connId = mnodeCreateConn(connInfo.user, connInfo.clientIp, connInfo.clientPort);
-  if (connId == 0) code = terrno;
+  SConnObj *pConn = mnodeCreateConn(connInfo.user, connInfo.clientIp, connInfo.clientPort);
+  if (pConn == NULL) {
+    code = terrno;
+  } else {
+    pConnectRsp->connId = htonl(pConn->connId);
+    mnodeReleaseConn(pConn);
+  }
 
   sprintf(pConnectRsp->acctId, "%x", pAcct->acctId);
   strcpy(pConnectRsp->serverVersion, version);
   pConnectRsp->writeAuth = pUser->writeAuth;
   pConnectRsp->superAuth = pUser->superAuth;
-  pConnectRsp->connId = htonl(connId);
   
   mnodeGetMnodeIpSetForShell(&pConnectRsp->ipList);
 

tests/script/sh/deploy.sh

@@ -111,7 +111,7 @@ echo "serverPort          ${NODE}"          >> $TAOS_CFG
 echo "dataDir             $DATA_DIR"      >> $TAOS_CFG
 echo "logDir              $LOG_DIR"       >> $TAOS_CFG
 echo "dDebugFlag          135"            >> $TAOS_CFG
-echo "mDebugFlag          135"            >> $TAOS_CFG
+echo "mDebugFlag          199"            >> $TAOS_CFG
 echo "sdbDebugFlag        135"            >> $TAOS_CFG
 echo "rpcDebugFlag        135"            >> $TAOS_CFG
 echo "tmrDebugFlag        131"            >> $TAOS_CFG