dndTransport.c 18.1 KB
Newer Older
S
Shengliang Guan 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23
/*
 * Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
 *
 * This program is free software: you can use, redistribute, and/or modify
 * it under the terms of the GNU Affero General Public License, version 3
 * or later ("AGPL"), as published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE.
 *
 * You should have received a copy of the GNU Affero General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */

/* this file is mainly responsible for the communication between DNODEs. Each
 * dnode works as both server and client. Dnode may send status, grant, config
 * messages to mnode, mnode may send create/alter/drop table/vnode messages
 * to dnode. All theses messages are handled from here
 */

#define _DEFAULT_SOURCE
#include "dndTransport.h"
S
Shengliang Guan 已提交
24
#include "dndMgmt.h"
S
Shengliang Guan 已提交
25 26 27
#include "dndMnode.h"
#include "dndVnodes.h"

L
Liu Jicong 已提交
28 29
#define INTERNAL_USER   "_dnd"
#define INTERNAL_CKEY   "_key"
S
Shengliang 已提交
30
#define INTERNAL_SECRET "_pwd"
S
Shengliang Guan 已提交
31

S
Shengliang Guan 已提交
32
static void dndInitMsgFp(STransMgmt *pMgmt) {
S
Shengliang Guan 已提交
33 34 35 36 37 38 39
  // Requests handled by DNODE
  pMgmt->msgFp[TMSG_INDEX(TDMT_DND_CREATE_MNODE)] = dndProcessMgmtMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_DND_CREATE_MNODE_RSP)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_DND_ALTER_MNODE)] = dndProcessMgmtMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_DND_ALTER_MNODE_RSP)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_DND_DROP_MNODE)] = dndProcessMgmtMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_DND_DROP_MNODE_RSP)] = dndProcessMnodeWriteMsg;
S
Shengliang Guan 已提交
40 41 42 43 44 45 46 47 48 49 50 51
  pMgmt->msgFp[TMSG_INDEX(TDMT_DND_CREATE_QNODE)] = dndProcessMgmtMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_DND_CREATE_QNODE_RSP)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_DND_DROP_QNODE)] = dndProcessMgmtMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_DND_DROP_QNODE_RSP)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_DND_CREATE_SNODE)] = dndProcessMgmtMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_DND_CREATE_SNODE_RSP)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_DND_DROP_SNODE)] = dndProcessMgmtMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_DND_DROP_SNODE_RSP)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_DND_CREATE_BNODE)] = dndProcessMgmtMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_DND_CREATE_BNODE_RSP)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_DND_DROP_BNODE)] = dndProcessMgmtMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_DND_DROP_BNODE_RSP)] = dndProcessMnodeWriteMsg;
S
Shengliang Guan 已提交
52 53 54 55 56 57 58 59 60 61 62 63 64
  pMgmt->msgFp[TMSG_INDEX(TDMT_DND_CREATE_VNODE)] = dndProcessMgmtMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_DND_CREATE_VNODE_RSP)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_DND_ALTER_VNODE)] = dndProcessMgmtMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_DND_ALTER_VNODE_RSP)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_DND_DROP_VNODE)] = dndProcessMgmtMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_DND_DROP_VNODE_RSP)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_DND_SYNC_VNODE)] = dndProcessMgmtMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_DND_SYNC_VNODE_RSP)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_DND_COMPACT_VNODE)] = dndProcessMgmtMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_DND_COMPACT_VNODE_RSP)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_DND_CONFIG_DNODE)] = dndProcessMgmtMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_DND_CONFIG_DNODE_RSP)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_DND_NETWORK_TEST)] = dndProcessMgmtMsg;
S
Shengliang Guan 已提交
65

S
Shengliang Guan 已提交
66
  // Requests handled by MNODE
67 68 69 70 71 72 73
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_CONNECT)] = dndProcessMnodeReadMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_CREATE_ACCT)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_ALTER_ACCT)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_DROP_ACCT)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_CREATE_USER)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_ALTER_USER)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_DROP_USER)] = dndProcessMnodeWriteMsg;
S
Shengliang Guan 已提交
74
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_GET_USER_AUTH)] = dndProcessMnodeReadMsg;
75 76 77 78 79
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_CREATE_DNODE)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_CONFIG_DNODE)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_DROP_DNODE)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_CREATE_MNODE)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_DROP_MNODE)] = dndProcessMnodeWriteMsg;
S
Shengliang Guan 已提交
80 81 82 83 84 85
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_CREATE_QNODE)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_DROP_QNODE)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_CREATE_SNODE)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_DROP_SNODE)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_CREATE_BNODE)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_DROP_BNODE)] = dndProcessMnodeWriteMsg;
86 87 88 89 90 91
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_CREATE_DB)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_DROP_DB)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_USE_DB)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_ALTER_DB)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_SYNC_DB)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_COMPACT_DB)] = dndProcessMnodeWriteMsg;
S
Shengliang 已提交
92 93 94
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_CREATE_FUNC)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_RETRIEVE_FUNC)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_DROP_FUNC)] = dndProcessMnodeWriteMsg;
95 96 97
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_CREATE_STB)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_ALTER_STB)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_DROP_STB)] = dndProcessMnodeWriteMsg;
D
dapan1121 已提交
98
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_TABLE_META)] = dndProcessMnodeReadMsg;
99 100 101
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_VGROUP_LIST)] = dndProcessMnodeReadMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_KILL_QUERY)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_KILL_CONN)] = dndProcessMnodeWriteMsg;
L
Liu Jicong 已提交
102
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_HEARTBEAT)] = dndProcessMnodeWriteMsg;
103 104
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_SHOW)] = dndProcessMnodeReadMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_SHOW_RETRIEVE)] = dndProcessMnodeReadMsg;
S
Shengliang Guan 已提交
105
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_STATUS)] = dndProcessMnodeReadMsg;
S
Shengliang Guan 已提交
106
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_STATUS_RSP)] = dndProcessMgmtMsg;
107
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_KILL_TRANS)] = dndProcessMnodeWriteMsg;
S
Shengliang Guan 已提交
108 109 110 111 112 113 114
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_GRANT)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_GRANT_RSP)] = dndProcessMgmtMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_AUTH)] = dndProcessMnodeReadMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_AUTH_RSP)] = dndProcessMgmtMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_CREATE_TOPIC)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_ALTER_TOPIC)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_DROP_TOPIC)] = dndProcessMnodeWriteMsg;
L
Liu Jicong 已提交
115
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_SUBSCRIBE)] = dndProcessMnodeWriteMsg;
L
Liu Jicong 已提交
116
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_MQ_COMMIT_OFFSET)] = dndProcessMnodeWriteMsg;
L
Liu Jicong 已提交
117 118
  /*pMgmt->msgFp[TMSG_INDEX(TDMT_VND_SUBSCRIBE_RSP)] = dndProcessMnodeWriteMsg;*/
  pMgmt->msgFp[TMSG_INDEX(TDMT_VND_MQ_SET_CONN_RSP)] = dndProcessMnodeWriteMsg;
L
Liu Jicong 已提交
119
  pMgmt->msgFp[TMSG_INDEX(TDMT_VND_MQ_REB_RSP)] = dndProcessMnodeWriteMsg;
L
Liu Jicong 已提交
120
  pMgmt->msgFp[TMSG_INDEX(TDMT_MND_GET_SUB_EP)] = dndProcessMnodeReadMsg;
S
Shengliang Guan 已提交
121

S
Shengliang Guan 已提交
122 123 124
  // Requests handled by VNODE
  pMgmt->msgFp[TMSG_INDEX(TDMT_VND_SUBMIT)] = dndProcessVnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_VND_QUERY)] = dndProcessVnodeQueryMsg;
S
Shengliang 已提交
125
  pMgmt->msgFp[TMSG_INDEX(TDMT_VND_QUERY_CONTINUE)] = dndProcessVnodeQueryMsg;
S
Shengliang Guan 已提交
126
  pMgmt->msgFp[TMSG_INDEX(TDMT_VND_FETCH)] = dndProcessVnodeFetchMsg;
S
Shengliang 已提交
127
  pMgmt->msgFp[TMSG_INDEX(TDMT_VND_FETCH_RSP)] = dndProcessVnodeFetchMsg;
S
Shengliang Guan 已提交
128 129
  pMgmt->msgFp[TMSG_INDEX(TDMT_VND_ALTER_TABLE)] = dndProcessVnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_VND_UPDATE_TAG_VAL)] = dndProcessVnodeWriteMsg;
D
dapan1121 已提交
130 131
  pMgmt->msgFp[TMSG_INDEX(TDMT_VND_TABLE_META)] = dndProcessVnodeFetchMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_VND_TABLES_META)] = dndProcessVnodeFetchMsg;
S
Shengliang Guan 已提交
132 133 134 135 136 137 138 139 140
  pMgmt->msgFp[TMSG_INDEX(TDMT_VND_MQ_CONSUME)] = dndProcessVnodeQueryMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_VND_MQ_QUERY)] = dndProcessVnodeQueryMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_VND_MQ_CONNECT)] = dndProcessVnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_VND_MQ_DISCONNECT)] = dndProcessVnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_VND_MQ_SET_CUR)] = dndProcessVnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_VND_RES_READY)] = dndProcessVnodeFetchMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_VND_TASKS_STATUS)] = dndProcessVnodeFetchMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_VND_CANCEL_TASK)] = dndProcessVnodeFetchMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_VND_DROP_TASK)] = dndProcessVnodeFetchMsg;
141 142 143 144 145 146
  pMgmt->msgFp[TMSG_INDEX(TDMT_VND_CREATE_STB)] = dndProcessVnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_VND_CREATE_STB_RSP)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_VND_ALTER_STB)] = dndProcessVnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_VND_ALTER_STB_RSP)] = dndProcessMnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_VND_DROP_STB)] = dndProcessVnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_VND_DROP_STB_RSP)] = dndProcessMnodeWriteMsg;
S
Shengliang Guan 已提交
147 148 149
  pMgmt->msgFp[TMSG_INDEX(TDMT_VND_CREATE_TABLE)] = dndProcessVnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_VND_ALTER_TABLE)] = dndProcessVnodeWriteMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_VND_DROP_TABLE)] = dndProcessVnodeWriteMsg;
H
Haojun Liao 已提交
150 151
  pMgmt->msgFp[TMSG_INDEX(TDMT_VND_SHOW_TABLES)] = dndProcessVnodeFetchMsg;
  pMgmt->msgFp[TMSG_INDEX(TDMT_VND_SHOW_TABLES_FETCH)] = dndProcessVnodeFetchMsg;
L
Liu Jicong 已提交
152
  pMgmt->msgFp[TMSG_INDEX(TDMT_VND_MQ_SET_CONN)] = dndProcessVnodeWriteMsg;
L
Liu Jicong 已提交
153
  pMgmt->msgFp[TMSG_INDEX(TDMT_VND_MQ_REB)] = dndProcessVnodeWriteMsg;
L
Liu Jicong 已提交
154
  pMgmt->msgFp[TMSG_INDEX(TDMT_VND_MQ_SET_CUR)] = dndProcessVnodeFetchMsg;
L
Liu Jicong 已提交
155
  pMgmt->msgFp[TMSG_INDEX(TDMT_VND_CONSUME)] = dndProcessVnodeFetchMsg;
S
Shengliang Guan 已提交
156 157
}

S
Shengliang Guan 已提交
158
static void dndProcessResponse(void *parent, SRpcMsg *pRsp, SEpSet *pEpSet) {
L
Liu Jicong 已提交
159
  SDnode     *pDnode = parent;
S
Shengliang Guan 已提交
160
  STransMgmt *pMgmt = &pDnode->tmgmt;
S
Shengliang Guan 已提交
161

S
Shengliang Guan 已提交
162
  tmsg_t msgType = pRsp->msgType;
S
Shengliang Guan 已提交
163 164

  if (dndGetStat(pDnode) == DND_STAT_STOPPED) {
S
Shengliang Guan 已提交
165
    if (pRsp == NULL || pRsp->pCont == NULL) return;
S
Shengliang 已提交
166
    dTrace("RPC %p, rsp:%s ignored since dnode exiting, app:%p", pRsp->handle, TMSG_INFO(msgType), pRsp->ahandle);
S
Shengliang Guan 已提交
167
    rpcFreeCont(pRsp->pCont);
S
Shengliang Guan 已提交
168 169 170
    return;
  }

171
  DndMsgFp fp = pMgmt->msgFp[TMSG_INDEX(msgType)];
S
Shengliang Guan 已提交
172
  if (fp != NULL) {
S
Shengliang 已提交
173 174
    dTrace("RPC %p, rsp:%s will be processed, code:0x%x app:%p", pRsp->handle, TMSG_INFO(msgType), pRsp->code & 0XFFFF,
           pRsp->ahandle);
S
Shengliang Guan 已提交
175
    (*fp)(pDnode, pRsp, pEpSet);
S
Shengliang Guan 已提交
176
  } else {
S
Shengliang 已提交
177
    dError("RPC %p, rsp:%s not processed, app:%p", pRsp->handle, TMSG_INFO(msgType), pRsp->ahandle);
S
Shengliang Guan 已提交
178
    rpcFreeCont(pRsp->pCont);
S
Shengliang Guan 已提交
179 180 181 182
  }
}

static int32_t dndInitClient(SDnode *pDnode) {
S
Shengliang Guan 已提交
183
  STransMgmt *pMgmt = &pDnode->tmgmt;
S
Shengliang Guan 已提交
184 185 186

  SRpcInit rpcInit;
  memset(&rpcInit, 0, sizeof(rpcInit));
S
Shengliang 已提交
187
  rpcInit.label = "D-C";
S
Shengliang Guan 已提交
188 189
  rpcInit.numOfThreads = 1;
  rpcInit.cfp = dndProcessResponse;
S
Shengliang Guan 已提交
190
  rpcInit.sessions = 1024;
S
Shengliang Guan 已提交
191
  rpcInit.connType = TAOS_CONN_CLIENT;
S
Shengliang Guan 已提交
192
  rpcInit.idleTime = tsShellActivityTimer * 1000;
S
Shengliang Guan 已提交
193 194
  rpcInit.user = INTERNAL_USER;
  rpcInit.ckey = INTERNAL_CKEY;
S
Shengliang Guan 已提交
195
  rpcInit.spi = 1;
196
  rpcInit.parent = pDnode;
dengyihao's avatar
fix bug  
dengyihao 已提交
197
  rpcInit.noPool = true;
S
Shengliang Guan 已提交
198

S
Shengliang Guan 已提交
199 200
  char pass[TSDB_PASSWORD_LEN + 1] = {0};
  taosEncryptPass_c((uint8_t *)(INTERNAL_SECRET), strlen(INTERNAL_SECRET), pass);
S
Shengliang Guan 已提交
201 202
  rpcInit.secret = pass;

S
Shengliang Guan 已提交
203 204
  pMgmt->clientRpc = rpcOpen(&rpcInit);
  if (pMgmt->clientRpc == NULL) {
S
Shengliang 已提交
205
    dError("failed to init dnode rpc client");
S
Shengliang Guan 已提交
206 207 208
    return -1;
  }

209
  dDebug("dnode rpc client is initialized");
S
Shengliang Guan 已提交
210 211 212 213
  return 0;
}

static void dndCleanupClient(SDnode *pDnode) {
S
Shengliang Guan 已提交
214
  STransMgmt *pMgmt = &pDnode->tmgmt;
S
Shengliang Guan 已提交
215 216 217
  if (pMgmt->clientRpc) {
    rpcClose(pMgmt->clientRpc);
    pMgmt->clientRpc = NULL;
218
    dDebug("dnode rpc client is closed");
S
Shengliang Guan 已提交
219 220 221
  }
}

S
Shengliang Guan 已提交
222
static void dndProcessRequest(void *param, SRpcMsg *pReq, SEpSet *pEpSet) {
L
Liu Jicong 已提交
223
  SDnode     *pDnode = param;
S
Shengliang Guan 已提交
224
  STransMgmt *pMgmt = &pDnode->tmgmt;
S
Shengliang Guan 已提交
225

S
Shengliang Guan 已提交
226
  tmsg_t msgType = pReq->msgType;
H
Hongze Cheng 已提交
227
  if (msgType == TDMT_DND_NETWORK_TEST) {
S
Shengliang 已提交
228
    dTrace("RPC %p, network test req will be processed, app:%p", pReq->handle, pReq->ahandle);
S
Shengliang Guan 已提交
229
    dndProcessStartupReq(pDnode, pReq);
S
Shengliang Guan 已提交
230 231 232 233
    return;
  }

  if (dndGetStat(pDnode) == DND_STAT_STOPPED) {
S
Shengliang 已提交
234 235
    dError("RPC %p, req:%s ignored since dnode exiting, app:%p", pReq->handle, TMSG_INFO(msgType), pReq->ahandle);
    SRpcMsg rspMsg = {.handle = pReq->handle, .code = TSDB_CODE_DND_OFFLINE, .ahandle = pReq->ahandle};
S
Shengliang Guan 已提交
236
    rpcSendResponse(&rspMsg);
S
Shengliang Guan 已提交
237
    rpcFreeCont(pReq->pCont);
S
Shengliang Guan 已提交
238 239
    return;
  } else if (dndGetStat(pDnode) != DND_STAT_RUNNING) {
S
Shengliang 已提交
240 241
    dError("RPC %p, req:%s ignored since dnode not running, app:%p", pReq->handle, TMSG_INFO(msgType), pReq->ahandle);
    SRpcMsg rspMsg = {.handle = pReq->handle, .code = TSDB_CODE_APP_NOT_READY, .ahandle = pReq->ahandle};
S
Shengliang Guan 已提交
242
    rpcSendResponse(&rspMsg);
S
Shengliang Guan 已提交
243
    rpcFreeCont(pReq->pCont);
S
Shengliang Guan 已提交
244 245 246
    return;
  }

S
Shengliang Guan 已提交
247
  if (pReq->pCont == NULL) {
S
Shengliang 已提交
248 249
    dTrace("RPC %p, req:%s not processed since its empty, app:%p", pReq->handle, TMSG_INFO(msgType), pReq->ahandle);
    SRpcMsg rspMsg = {.handle = pReq->handle, .code = TSDB_CODE_DND_INVALID_MSG_LEN, .ahandle = pReq->ahandle};
S
Shengliang Guan 已提交
250 251 252 253
    rpcSendResponse(&rspMsg);
    return;
  }

254
  DndMsgFp fp = pMgmt->msgFp[TMSG_INDEX(msgType)];
S
Shengliang Guan 已提交
255
  if (fp != NULL) {
S
Shengliang 已提交
256
    dTrace("RPC %p, req:%s will be processed, app:%p", pReq->handle, TMSG_INFO(msgType), pReq->ahandle);
S
Shengliang Guan 已提交
257
    (*fp)(pDnode, pReq, pEpSet);
S
Shengliang Guan 已提交
258
  } else {
S
Shengliang 已提交
259 260
    dError("RPC %p, req:%s not processed since no handle, app:%p", pReq->handle, TMSG_INFO(msgType), pReq->ahandle);
    SRpcMsg rspMsg = {.handle = pReq->handle, .code = TSDB_CODE_MSG_NOT_PROCESSED, .ahandle = pReq->ahandle};
S
Shengliang Guan 已提交
261
    rpcSendResponse(&rspMsg);
S
Shengliang Guan 已提交
262
    rpcFreeCont(pReq->pCont);
S
Shengliang Guan 已提交
263 264 265 266
  }
}

static void dndSendMsgToMnodeRecv(SDnode *pDnode, SRpcMsg *pRpcMsg, SRpcMsg *pRpcRsp) {
S
Shengliang Guan 已提交
267
  STransMgmt *pMgmt = &pDnode->tmgmt;
S
Shengliang Guan 已提交
268 269 270 271 272 273

  SEpSet epSet = {0};
  dndGetMnodeEpSet(pDnode, &epSet);
  rpcSendRecv(pMgmt->clientRpc, &epSet, pRpcMsg, pRpcRsp);
}

S
Shengliang Guan 已提交
274
static int32_t dndAuthInternalReq(SDnode *pDnode, char *user, char *spi, char *encrypt, char *secret, char *ckey) {
S
Shengliang Guan 已提交
275
  if (strcmp(user, INTERNAL_USER) == 0) {
S
Shengliang Guan 已提交
276 277
    char pass[TSDB_PASSWORD_LEN + 1] = {0};
    taosEncryptPass_c((uint8_t *)(INTERNAL_SECRET), strlen(INTERNAL_SECRET), pass);
278
    memcpy(secret, pass, TSDB_PASSWORD_LEN);
S
Shengliang Guan 已提交
279
    *spi = 1;
S
Shengliang Guan 已提交
280 281 282 283
    *encrypt = 0;
    *ckey = 0;
    return 0;
  } else if (strcmp(user, TSDB_NETTEST_USER) == 0) {
S
Shengliang Guan 已提交
284 285
    char pass[TSDB_PASSWORD_LEN + 1] = {0};
    taosEncryptPass_c((uint8_t *)(TSDB_NETTEST_USER), strlen(TSDB_NETTEST_USER), pass);
286
    memcpy(secret, pass, TSDB_PASSWORD_LEN);
S
Shengliang Guan 已提交
287
    *spi = 1;
S
Shengliang Guan 已提交
288 289 290 291 292 293 294 295
    *encrypt = 0;
    *ckey = 0;
    return 0;
  } else {
    return -1;
  }
}

S
Shengliang Guan 已提交
296 297 298
static int32_t dndRetrieveUserAuthInfo(void *parent, char *user, char *spi, char *encrypt, char *secret, char *ckey) {
  SDnode *pDnode = parent;

S
Shengliang Guan 已提交
299
  if (dndAuthInternalReq(parent, user, spi, encrypt, secret, ckey) == 0) {
S
Shengliang 已提交
300
    dTrace("user:%s, get auth from mnode, spi:%d encrypt:%d", user, *spi, *encrypt);
S
Shengliang Guan 已提交
301 302 303 304
    return 0;
  }

  if (dndGetUserAuthFromMnode(pDnode, user, spi, encrypt, secret, ckey) == 0) {
S
Shengliang 已提交
305
    dTrace("user:%s, get auth from mnode, spi:%d encrypt:%d", user, *spi, *encrypt);
S
Shengliang Guan 已提交
306 307 308 309
    return 0;
  }

  if (terrno != TSDB_CODE_APP_NOT_READY) {
S
Shengliang 已提交
310
    dTrace("failed to get user auth from mnode since %s", terrstr());
S
Shengliang Guan 已提交
311
    return -1;
S
Shengliang Guan 已提交
312 313
  }

S
Shengliang Guan 已提交
314 315 316
  SAuthReq authReq = {0};
  tstrncpy(authReq.user, user, TSDB_USER_LEN);
  int32_t contLen = tSerializeSAuthReq(NULL, 0, &authReq);
L
Liu Jicong 已提交
317
  void   *pReq = rpcMallocCont(contLen);
S
Shengliang Guan 已提交
318
  tSerializeSAuthReq(pReq, contLen, &authReq);
S
Shengliang Guan 已提交
319

S
Shengliang Guan 已提交
320
  SRpcMsg rpcMsg = {.pCont = pReq, .contLen = contLen, .msgType = TDMT_MND_AUTH, .ahandle = (void *)9528};
S
Shengliang Guan 已提交
321
  SRpcMsg rpcRsp = {0};
S
Shengliang Guan 已提交
322
  dTrace("user:%s, send user auth req to other mnodes, spi:%d encrypt:%d", user, authReq.spi, authReq.encrypt);
S
Shengliang Guan 已提交
323 324 325 326
  dndSendMsgToMnodeRecv(pDnode, &rpcMsg, &rpcRsp);

  if (rpcRsp.code != 0) {
    terrno = rpcRsp.code;
S
Shengliang Guan 已提交
327
    dError("user:%s, failed to get user auth from other mnodes since %s", user, terrstr());
S
Shengliang Guan 已提交
328
  } else {
S
Shengliang Guan 已提交
329 330 331 332 333 334 335 336
    SAuthRsp authRsp = {0};
    tDeserializeSAuthReq(rpcRsp.pCont, rpcRsp.contLen, &authRsp);
    memcpy(secret, authRsp.secret, TSDB_PASSWORD_LEN);
    memcpy(ckey, authRsp.ckey, TSDB_PASSWORD_LEN);
    *spi = authRsp.spi;
    *encrypt = authRsp.encrypt;
    dTrace("user:%s, success to get user auth from other mnodes, spi:%d encrypt:%d", user, authRsp.spi,
           authRsp.encrypt);
S
Shengliang Guan 已提交
337 338 339 340 341 342 343
  }

  rpcFreeCont(rpcRsp.pCont);
  return rpcRsp.code;
}

static int32_t dndInitServer(SDnode *pDnode) {
S
Shengliang Guan 已提交
344
  STransMgmt *pMgmt = &pDnode->tmgmt;
S
Shengliang Guan 已提交
345 346
  dndInitMsgFp(pMgmt);

S
config  
Shengliang Guan 已提交
347
  int32_t numOfThreads = (int32_t)((tsNumOfCores * tsNumOfThreadsPerCore) / 2.0);
S
Shengliang Guan 已提交
348 349 350 351 352 353
  if (numOfThreads < 1) {
    numOfThreads = 1;
  }

  SRpcInit rpcInit;
  memset(&rpcInit, 0, sizeof(rpcInit));
S
Shengliang Guan 已提交
354
  rpcInit.localPort = pDnode->cfg.serverPort;
S
Shengliang 已提交
355
  rpcInit.label = "D-S";
S
Shengliang Guan 已提交
356 357
  rpcInit.numOfThreads = numOfThreads;
  rpcInit.cfp = dndProcessRequest;
S
Shengliang Guan 已提交
358
  rpcInit.sessions = tsMaxShellConns;
S
Shengliang Guan 已提交
359
  rpcInit.connType = TAOS_CONN_SERVER;
S
Shengliang Guan 已提交
360
  rpcInit.idleTime = tsShellActivityTimer * 1000;
S
Shengliang Guan 已提交
361
  rpcInit.afp = dndRetrieveUserAuthInfo;
362
  rpcInit.parent = pDnode;
S
Shengliang Guan 已提交
363 364 365

  pMgmt->serverRpc = rpcOpen(&rpcInit);
  if (pMgmt->serverRpc == NULL) {
S
Shengliang 已提交
366
    dError("failed to init dnode rpc server");
S
Shengliang Guan 已提交
367 368 369
    return -1;
  }

370
  dDebug("dnode rpc server is initialized");
S
Shengliang Guan 已提交
371 372 373 374
  return 0;
}

static void dndCleanupServer(SDnode *pDnode) {
S
Shengliang Guan 已提交
375
  STransMgmt *pMgmt = &pDnode->tmgmt;
S
Shengliang Guan 已提交
376 377 378
  if (pMgmt->serverRpc) {
    rpcClose(pMgmt->serverRpc);
    pMgmt->serverRpc = NULL;
379
    dDebug("dnode rpc server is closed");
S
Shengliang Guan 已提交
380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396
  }
}

int32_t dndInitTrans(SDnode *pDnode) {
  if (dndInitClient(pDnode) != 0) {
    return -1;
  }

  if (dndInitServer(pDnode) != 0) {
    return -1;
  }

  dInfo("dnode-transport is initialized");
  return 0;
}

void dndCleanupTrans(SDnode *pDnode) {
397
  dInfo("dnode-transport start to clean up");
S
Shengliang Guan 已提交
398 399 400 401 402
  dndCleanupServer(pDnode);
  dndCleanupClient(pDnode);
  dInfo("dnode-transport is cleaned up");
}

S
Shengliang Guan 已提交
403
int32_t dndSendReqToDnode(SDnode *pDnode, SEpSet *pEpSet, SRpcMsg *pReq) {
S
Shengliang Guan 已提交
404
  STransMgmt *pMgmt = &pDnode->tmgmt;
S
Shengliang Guan 已提交
405 406 407 408 409
  if (pMgmt->clientRpc == NULL) {
    terrno = TSDB_CODE_DND_OFFLINE;
    return -1;
  }

S
Shengliang Guan 已提交
410
  rpcSendRequest(pMgmt->clientRpc, pEpSet, pReq, NULL);
S
Shengliang Guan 已提交
411
  return 0;
S
Shengliang Guan 已提交
412 413
}

S
Shengliang Guan 已提交
414
int32_t dndSendReqToMnode(SDnode *pDnode, SRpcMsg *pReq) {
S
Shengliang Guan 已提交
415 416
  SEpSet epSet = {0};
  dndGetMnodeEpSet(pDnode, &epSet);
S
Shengliang Guan 已提交
417
  return dndSendReqToDnode(pDnode, &epSet, pReq);
D
dapan1121 已提交
418
}