taosdata / TDengine / #3621

coverage: 2.037% (-61.5%) from 63.573%
#3621

push

travis-ci

web-flow 
Merge pull request #29874 from taosdata/merge/mainto3.0

merge: from main to 3.0 branch

4357 of 287032 branches covered (1.52%)

Branch coverage included in aggregate %.

0 of 174 new or added lines in 18 files covered. (0.0%)

213359 existing lines in 469 files now uncovered.

7260 of 283369 relevant lines covered (2.56%)

23737.72 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

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









16



#define _DEFAULT_SOURCE










17



#include "mndAnode.h"










18



#include "audit.h"










19



#include "mndDnode.h"










20



#include "mndPrivilege.h"










21



#include "mndShow.h"










22



#include "mndTrans.h"










23



#include "mndUser.h"










24



#include "tanalytics.h"










25



#include "tjson.h"










26












27



#ifdef USE_ANALYTICS










28












29



#define TSDB_ANODE_VER_NUMBER   1










30



#define TSDB_ANODE_RESERVE_SIZE 64










31












32



static SSdbRaw *mndAnodeActionEncode(SAnodeObj *pObj);










33



static SSdbRow *mndAnodeActionDecode(SSdbRaw *pRaw);










34



static int32_t  mndAnodeActionInsert(SSdb *pSdb, SAnodeObj *pObj);










35



static int32_t  mndAnodeActionUpdate(SSdb *pSdb, SAnodeObj *pOld, SAnodeObj *pNew);










36



static int32_t  mndAnodeActionDelete(SSdb *pSdb, SAnodeObj *pObj);










37



static int32_t  mndProcessCreateAnodeReq(SRpcMsg *pReq);










38



static int32_t  mndProcessUpdateAnodeReq(SRpcMsg *pReq);










39



static int32_t  mndProcessDropAnodeReq(SRpcMsg *pReq);










40



static int32_t  mndProcessAnalAlgoReq(SRpcMsg *pReq);










41



static int32_t  mndRetrieveAnodes(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows);










42



static void     mndCancelGetNextAnode(SMnode *pMnode, void *pIter);










43



static int32_t  mndRetrieveAnodesFull(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows);










44



static void     mndCancelGetNextAnodeFull(SMnode *pMnode, void *pIter);










45



static int32_t  mndGetAnodeAlgoList(const char *url, SAnodeObj *pObj);










46



static int32_t  mndGetAnodeStatus(SAnodeObj *pObj, char *status, int32_t statusLen);










47













UNCOV

48



int32_t mndInitAnode(SMnode *pMnode) {




×








UNCOV

49



  SSdbTable table = {




×







50



      .sdbType = SDB_ANODE,










51



      .keyType = SDB_KEY_INT32,










52



      .encodeFp = (SdbEncodeFp)mndAnodeActionEncode,










53



      .decodeFp = (SdbDecodeFp)mndAnodeActionDecode,










54



      .insertFp = (SdbInsertFp)mndAnodeActionInsert,










55



      .updateFp = (SdbUpdateFp)mndAnodeActionUpdate,










56



      .deleteFp = (SdbDeleteFp)mndAnodeActionDelete,










57



  };










58













UNCOV

59



  mndSetMsgHandle(pMnode, TDMT_MND_CREATE_ANODE, mndProcessCreateAnodeReq);




×








UNCOV

60



  mndSetMsgHandle(pMnode, TDMT_MND_UPDATE_ANODE, mndProcessUpdateAnodeReq);




×








UNCOV

61



  mndSetMsgHandle(pMnode, TDMT_MND_DROP_ANODE, mndProcessDropAnodeReq);




×








UNCOV

62



  mndSetMsgHandle(pMnode, TDMT_MND_RETRIEVE_ANAL_ALGO, mndProcessAnalAlgoReq);




×







63













UNCOV

64



  mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_ANODE, mndRetrieveAnodes);




×








UNCOV

65



  mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_ANODE, mndCancelGetNextAnode);




×








UNCOV

66



  mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_ANODE_FULL, mndRetrieveAnodesFull);




×








UNCOV

67



  mndAddShowFreeIterHandle(pMnode, TSDB_MGMT_TABLE_ANODE_FULL, mndCancelGetNextAnodeFull);




×







68













UNCOV

69



  return sdbSetTable(pMnode->pSdb, table);




×







70



}










71













UNCOV

72



void mndCleanupAnode(SMnode *pMnode) {}




×







73













UNCOV

74



SAnodeObj *mndAcquireAnode(SMnode *pMnode, int32_t anodeId) {




×








UNCOV

75



  SAnodeObj *pObj = sdbAcquire(pMnode->pSdb, SDB_ANODE, &anodeId);




×








UNCOV

76



  if (pObj == NULL && terrno == TSDB_CODE_SDB_OBJ_NOT_THERE) {





×








UNCOV

77



    terrno = TSDB_CODE_MND_ANODE_NOT_EXIST;




×







78



  }











UNCOV

79



  return pObj;




×







80



}










81













UNCOV

82



void mndReleaseAnode(SMnode *pMnode, SAnodeObj *pObj) {




×








UNCOV

83



  SSdb *pSdb = pMnode->pSdb;




×








UNCOV

84



  sdbRelease(pSdb, pObj);




×








UNCOV

85



}




×







86













UNCOV

87



static SSdbRaw *mndAnodeActionEncode(SAnodeObj *pObj) {




×








UNCOV

88



  int32_t code = 0;




×








UNCOV

89



  int32_t lino = 0;




×








UNCOV

90



  terrno = TSDB_CODE_OUT_OF_MEMORY;




×







91













UNCOV

92



  int32_t rawDataLen = sizeof(SAnodeObj) + TSDB_ANODE_RESERVE_SIZE + pObj->urlLen;




×








UNCOV

93



  for (int32_t t = 0; t < pObj->numOfAlgos; ++t) {





×








UNCOV

94



    SArray *algos = pObj->algos[t];




×








UNCOV

95



    for (int32_t a = 0; a < (int32_t)taosArrayGetSize(algos); ++a) {





×








UNCOV

96



      SAnodeAlgo *algo = taosArrayGet(algos, a);




×








UNCOV

97



      rawDataLen += (2 * sizeof(int32_t) + algo->nameLen);




×







98



    }











UNCOV

99



    rawDataLen += sizeof(int32_t);




×







100



  }










101













UNCOV

102



  SSdbRaw *pRaw = sdbAllocRaw(SDB_ANODE, TSDB_ANODE_VER_NUMBER, rawDataLen);




×








UNCOV

103



  if (pRaw == NULL) goto _OVER;





×







104













UNCOV

105



  int32_t dataPos = 0;




×








UNCOV

106



  SDB_SET_INT32(pRaw, dataPos, pObj->id, _OVER)





×








UNCOV

107



  SDB_SET_INT64(pRaw, dataPos, pObj->createdTime, _OVER)





×








UNCOV

108



  SDB_SET_INT64(pRaw, dataPos, pObj->updateTime, _OVER)





×








UNCOV

109



  SDB_SET_INT32(pRaw, dataPos, pObj->version, _OVER)





×








UNCOV

110



  SDB_SET_INT32(pRaw, dataPos, pObj->urlLen, _OVER)





×








UNCOV

111



  SDB_SET_BINARY(pRaw, dataPos, pObj->url, pObj->urlLen, _OVER)





×








UNCOV

112



  SDB_SET_INT32(pRaw, dataPos, pObj->numOfAlgos, _OVER)





×








UNCOV

113



  for (int32_t i = 0; i < pObj->numOfAlgos; ++i) {





×








UNCOV

114



    SArray *algos = pObj->algos[i];




×








UNCOV

115



    SDB_SET_INT32(pRaw, dataPos, (int32_t)taosArrayGetSize(algos), _OVER)





×








UNCOV

116



    for (int32_t j = 0; j < (int32_t)taosArrayGetSize(algos); ++j) {





×








UNCOV

117



      SAnodeAlgo *algo = taosArrayGet(algos, j);




×








UNCOV

118



      SDB_SET_INT32(pRaw, dataPos, algo->nameLen, _OVER)





×








UNCOV

119



      SDB_SET_BINARY(pRaw, dataPos, algo->name, algo->nameLen, _OVER)





×








UNCOV

120



      SDB_SET_INT32(pRaw, dataPos, 0, _OVER)  // reserved





×







121



    }










122



  }










123













UNCOV

124



  SDB_SET_RESERVE(pRaw, dataPos, TSDB_ANODE_RESERVE_SIZE, _OVER)





×







125













UNCOV

126



  terrno = 0;




×







127













UNCOV

128



_OVER:




×








UNCOV

129



  if (terrno != 0) {





×







130



    mError("anode:%d, failed to encode to raw:%p since %s", pObj->id, pRaw, terrstr());





×







131



    sdbFreeRaw(pRaw);




×







132



    return NULL;




×







133



  }










134













UNCOV

135



  mTrace("anode:%d, encode to raw:%p, row:%p", pObj->id, pRaw, pObj);





×








UNCOV

136



  return pRaw;




×







137



}










138













UNCOV

139



static SSdbRow *mndAnodeActionDecode(SSdbRaw *pRaw) {




×








UNCOV

140



  int32_t code = 0;




×








UNCOV

141



  int32_t lino = 0;




×








UNCOV

142



  terrno = TSDB_CODE_OUT_OF_MEMORY;




×








UNCOV

143



  SSdbRow   *pRow = NULL;




×








UNCOV

144



  SAnodeObj *pObj = NULL;




×







145













UNCOV

146



  int8_t sver = 0;




×








UNCOV

147



  if (sdbGetRawSoftVer(pRaw, &sver) != 0) goto _OVER;





×







148













UNCOV

149



  if (sver != TSDB_ANODE_VER_NUMBER) {





×







150



    terrno = TSDB_CODE_SDB_INVALID_DATA_VER;




×







151



    goto _OVER;




×







152



  }










153













UNCOV

154



  pRow = sdbAllocRow(sizeof(SAnodeObj));




×








UNCOV

155



  if (pRow == NULL) goto _OVER;





×







156













UNCOV

157



  pObj = sdbGetRowObj(pRow);




×








UNCOV

158



  if (pObj == NULL) goto _OVER;





×







159













UNCOV

160



  int32_t dataPos = 0;




×








UNCOV

161



  SDB_GET_INT32(pRaw, dataPos, &pObj->id, _OVER)





×








UNCOV

162



  SDB_GET_INT64(pRaw, dataPos, &pObj->createdTime, _OVER)





×








UNCOV

163



  SDB_GET_INT64(pRaw, dataPos, &pObj->updateTime, _OVER)





×








UNCOV

164



  SDB_GET_INT32(pRaw, dataPos, &pObj->version, _OVER)





×








UNCOV

165



  SDB_GET_INT32(pRaw, dataPos, &pObj->urlLen, _OVER)





×







166













UNCOV

167



  if (pObj->urlLen > 0) {





×








UNCOV

168



    pObj->url = taosMemoryCalloc(pObj->urlLen, 1);





×








UNCOV

169



    if (pObj->url == NULL) goto _OVER;





×








UNCOV

170



    SDB_GET_BINARY(pRaw, dataPos, pObj->url, pObj->urlLen, _OVER)





×







171



  }










172













UNCOV

173



  SDB_GET_INT32(pRaw, dataPos, &pObj->numOfAlgos, _OVER)





×








UNCOV

174



  if (pObj->numOfAlgos > 0) {





×








UNCOV

175



    pObj->algos = taosMemoryCalloc(pObj->numOfAlgos, sizeof(SArray *));





×








UNCOV

176



    if (pObj->algos == NULL) {





×







177



      goto _OVER;




×







178



    }










179



  }










180













UNCOV

181



  for (int32_t i = 0; i < pObj->numOfAlgos; ++i) {





×








UNCOV

182



    int32_t numOfAlgos = 0;




×








UNCOV

183



    SDB_GET_INT32(pRaw, dataPos, &numOfAlgos, _OVER)





×







184













UNCOV

185



    pObj->algos[i] = taosArrayInit(2, sizeof(SAnodeAlgo));




×








UNCOV

186



    if (pObj->algos[i] == NULL) goto _OVER;





×







187













UNCOV

188



    for (int32_t j = 0; j < numOfAlgos; ++j) {





×








UNCOV

189



      SAnodeAlgo algoObj = {0};




×








UNCOV

190



      int32_t    reserved = 0;




×







191













UNCOV

192



      SDB_GET_INT32(pRaw, dataPos, &algoObj.nameLen, _OVER)





×








UNCOV

193



      if (algoObj.nameLen > 0) {





×








UNCOV

194



        algoObj.name = taosMemoryCalloc(algoObj.nameLen, 1);





×








UNCOV

195



        if (algoObj.name == NULL) goto _OVER;





×







196



      }










197













UNCOV

198



      SDB_GET_BINARY(pRaw, dataPos, algoObj.name, algoObj.nameLen, _OVER)





×








UNCOV

199



      SDB_GET_INT32(pRaw, dataPos, &reserved, _OVER);





×







200













UNCOV

201



      if (taosArrayPush(pObj->algos[i], &algoObj) == NULL) goto _OVER;





×







202



    }










203



  }










204













UNCOV

205



  SDB_GET_RESERVE(pRaw, dataPos, TSDB_ANODE_RESERVE_SIZE, _OVER)





×







206













UNCOV

207



  terrno = 0;




×







208













UNCOV

209



_OVER:




×








UNCOV

210



  if (terrno != 0) {





×







211



    mError("anode:%d, failed to decode from raw:%p since %s", pObj == NULL ? 0 : pObj->id, pRaw, terrstr());





×







212



    if (pObj != NULL) {





×







213



      taosMemoryFreeClear(pObj->url);





×







214



    }










215



    taosMemoryFreeClear(pRow);





×







216



    return NULL;




×







217



  }










218













UNCOV

219



  mTrace("anode:%d, decode from raw:%p, row:%p", pObj->id, pRaw, pObj);





×








UNCOV

220



  return pRow;




×







221



}










222













UNCOV

223



static void mndFreeAnode(SAnodeObj *pObj) {




×








UNCOV

224



  taosMemoryFreeClear(pObj->url);





×








UNCOV

225



  for (int32_t i = 0; i < pObj->numOfAlgos; ++i) {





×








UNCOV

226



    SArray *algos = pObj->algos[i];




×








UNCOV

227



    for (int32_t j = 0; j < (int32_t)taosArrayGetSize(algos); ++j) {





×








UNCOV

228



      SAnodeAlgo *algo = taosArrayGet(algos, j);




×








UNCOV

229



      taosMemoryFreeClear(algo->name);





×







230



    }











UNCOV

231



    taosArrayDestroy(algos);




×







232



  }











UNCOV

233



  taosMemoryFreeClear(pObj->algos);





×








UNCOV

234



}




×







235













UNCOV

236



static int32_t mndAnodeActionInsert(SSdb *pSdb, SAnodeObj *pObj) {




×








UNCOV

237



  mTrace("anode:%d, perform insert action, row:%p", pObj->id, pObj);





×








UNCOV

238



  return 0;




×







239



}










240













UNCOV

241



static int32_t mndAnodeActionDelete(SSdb *pSdb, SAnodeObj *pObj) {




×








UNCOV

242



  mTrace("anode:%d, perform delete action, row:%p", pObj->id, pObj);





×








UNCOV

243



  mndFreeAnode(pObj);




×








UNCOV

244



  return 0;




×







245



}










246













UNCOV

247



static int32_t mndAnodeActionUpdate(SSdb *pSdb, SAnodeObj *pOld, SAnodeObj *pNew) {




×








UNCOV

248



  mTrace("anode:%d, perform update action, old row:%p new row:%p", pOld->id, pOld, pNew);





×







249













UNCOV

250



  taosWLockLatch(&pOld->lock);




×








UNCOV

251



  int32_t numOfAlgos = pNew->numOfAlgos;




×








UNCOV

252



  void   *algos = pNew->algos;




×








UNCOV

253



  pNew->numOfAlgos = pOld->numOfAlgos;




×








UNCOV

254



  pNew->algos = pOld->algos;




×








UNCOV

255



  pOld->numOfAlgos = numOfAlgos;




×








UNCOV

256



  pOld->algos = algos;




×








UNCOV

257



  pOld->updateTime = pNew->updateTime;




×








UNCOV

258



  pOld->version = pNew->version;




×








UNCOV

259



  taosWUnLockLatch(&pOld->lock);




×








UNCOV

260



  return 0;




×







261



}










262













UNCOV

263



static int32_t mndSetCreateAnodeRedoLogs(STrans *pTrans, SAnodeObj *pObj) {




×








UNCOV

264



  int32_t  code = 0;




×








UNCOV

265



  SSdbRaw *pRedoRaw = mndAnodeActionEncode(pObj);




×








UNCOV

266



  if (pRedoRaw == NULL) {





×







267



    code = TSDB_CODE_MND_RETURN_VALUE_NULL;




×







268



    if (terrno != 0) code = terrno;





×







269



    TAOS_RETURN(code);




×







270



  }











UNCOV

271



  TAOS_CHECK_RETURN(mndTransAppendRedolog(pTrans, pRedoRaw));





×








UNCOV

272



  TAOS_CHECK_RETURN(sdbSetRawStatus(pRedoRaw, SDB_STATUS_CREATING));





×








UNCOV

273



  TAOS_RETURN(code);




×







274



}










275













UNCOV

276



static int32_t mndSetCreateAnodeUndoLogs(STrans *pTrans, SAnodeObj *pObj) {




×








UNCOV

277



  int32_t  code = 0;




×








UNCOV

278



  SSdbRaw *pUndoRaw = mndAnodeActionEncode(pObj);




×








UNCOV

279



  if (pUndoRaw == NULL) {





×







280



    code = TSDB_CODE_MND_RETURN_VALUE_NULL;




×







281



    if (terrno != 0) code = terrno;





×







282



    TAOS_RETURN(code);




×







283



  }











UNCOV

284



  TAOS_CHECK_RETURN(mndTransAppendUndolog(pTrans, pUndoRaw));





×








UNCOV

285



  TAOS_CHECK_RETURN(sdbSetRawStatus(pUndoRaw, SDB_STATUS_DROPPED));





×








UNCOV

286



  TAOS_RETURN(code);




×







287



}










288













UNCOV

289



static int32_t mndSetCreateAnodeCommitLogs(STrans *pTrans, SAnodeObj *pObj) {




×








UNCOV

290



  int32_t  code = 0;




×








UNCOV

291



  SSdbRaw *pCommitRaw = mndAnodeActionEncode(pObj);




×








UNCOV

292



  if (pCommitRaw == NULL) {





×







293



    code = TSDB_CODE_MND_RETURN_VALUE_NULL;




×







294



    if (terrno != 0) code = terrno;





×







295



    TAOS_RETURN(code);




×







296



  }











UNCOV

297



  TAOS_CHECK_RETURN(mndTransAppendCommitlog(pTrans, pCommitRaw));





×








UNCOV

298



  TAOS_CHECK_RETURN(sdbSetRawStatus(pCommitRaw, SDB_STATUS_READY));





×








UNCOV

299



  TAOS_RETURN(code);




×







300



}










301













UNCOV

302



static int32_t mndCreateAnode(SMnode *pMnode, SRpcMsg *pReq, SMCreateAnodeReq *pCreate) {




×








UNCOV

303



  int32_t code = -1;




×








UNCOV

304



  STrans *pTrans = NULL;




×







305













UNCOV

306



  SAnodeObj anodeObj = {0};




×








UNCOV

307



  anodeObj.id = sdbGetMaxId(pMnode->pSdb, SDB_ANODE);




×








UNCOV

308



  anodeObj.createdTime = taosGetTimestampMs();




×








UNCOV

309



  anodeObj.updateTime = anodeObj.createdTime;




×








UNCOV

310



  anodeObj.version = 0;




×








UNCOV

311



  anodeObj.urlLen = pCreate->urlLen;




×








UNCOV

312



  if (anodeObj.urlLen > TSDB_ANALYTIC_ANODE_URL_LEN) {





×







313



    code = TSDB_CODE_MND_ANODE_TOO_LONG_URL;




×







314



    goto _OVER;




×







315



  }










316













UNCOV

317



  anodeObj.url = taosMemoryCalloc(1, pCreate->urlLen);





×








UNCOV

318



  if (anodeObj.url == NULL) goto _OVER;





×








UNCOV

319



  (void)memcpy(anodeObj.url, pCreate->url, pCreate->urlLen);




×







320













UNCOV

321



  code = mndGetAnodeAlgoList(anodeObj.url, &anodeObj);




×








UNCOV

322



  if (code != 0) goto _OVER;





×







323













UNCOV

324



  pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_NOTHING, pReq, "create-anode");




×








UNCOV

325



  if (pTrans == NULL) {





×







326



    code = TSDB_CODE_MND_RETURN_VALUE_NULL;




×







327



    if (terrno != 0) code = terrno;





×







328



    goto _OVER;




×







329



  }











UNCOV

330



  mndTransSetSerial(pTrans);




×







331













UNCOV

332



  mInfo("trans:%d, used to create anode:%s as anode:%d", pTrans->id, pCreate->url, anodeObj.id);





×







333













UNCOV

334



  TAOS_CHECK_GOTO(mndSetCreateAnodeRedoLogs(pTrans, &anodeObj), NULL, _OVER);





×








UNCOV

335



  TAOS_CHECK_GOTO(mndSetCreateAnodeUndoLogs(pTrans, &anodeObj), NULL, _OVER);





×








UNCOV

336



  TAOS_CHECK_GOTO(mndSetCreateAnodeCommitLogs(pTrans, &anodeObj), NULL, _OVER);





×








UNCOV

337



  TAOS_CHECK_GOTO(mndTransPrepare(pMnode, pTrans), NULL, _OVER);





×







338













UNCOV

339



  code = 0;




×







340













UNCOV

341



_OVER:




×








UNCOV

342



  mndFreeAnode(&anodeObj);




×








UNCOV

343



  mndTransDrop(pTrans);




×








UNCOV

344



  TAOS_RETURN(code);




×







345



}










346













UNCOV

347



static SAnodeObj *mndAcquireAnodeByURL(SMnode *pMnode, char *url) {




×








UNCOV

348



  SSdb *pSdb = pMnode->pSdb;




×







349













UNCOV

350



  void *pIter = NULL;




×







351



  while (1) {




×








UNCOV

352



    SAnodeObj *pAnode = NULL;




×








UNCOV

353



    pIter = sdbFetch(pSdb, SDB_ANODE, pIter, (void **)&pAnode);




×








UNCOV

354



    if (pIter == NULL) break;





×







355












356



    if (strcasecmp(url, pAnode->url) == 0) {





×







357



      sdbCancelFetch(pSdb, pIter);




×







358



      return pAnode;




×







359



    }










360












361



    sdbRelease(pSdb, pAnode);




×







362



  }










363













UNCOV

364



  terrno = TSDB_CODE_MND_ANODE_NOT_EXIST;




×








UNCOV

365



  return NULL;




×







366



}










367













UNCOV

368



static int32_t mndProcessCreateAnodeReq(SRpcMsg *pReq) {




×








UNCOV

369



  SMnode          *pMnode = pReq->info.node;




×








UNCOV

370



  int32_t          code = -1;




×








UNCOV

371



  SAnodeObj       *pObj = NULL;




×








UNCOV

372



  SMCreateAnodeReq createReq = {0};




×







373













UNCOV

374



  TAOS_CHECK_GOTO(tDeserializeSMCreateAnodeReq(pReq->pCont, pReq->contLen, &createReq), NULL, _OVER);





×







375













UNCOV

376



  mInfo("anode:%s, start to create", createReq.url);





×








UNCOV

377



  TAOS_CHECK_GOTO(mndCheckOperPrivilege(pMnode, pReq->info.conn.user, MND_OPER_CREATE_ANODE), NULL, _OVER);





×







378













UNCOV

379



  pObj = mndAcquireAnodeByURL(pMnode, createReq.url);




×








UNCOV

380



  if (pObj != NULL) {





×







381



    code = TSDB_CODE_MND_ANODE_ALREADY_EXIST;




×







382



    goto _OVER;




×







383



  }










384













UNCOV

385



  code = mndCreateAnode(pMnode, pReq, &createReq);




×








UNCOV

386



  if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;





×







387













UNCOV

388



_OVER:




×








UNCOV

389



  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {





×








UNCOV

390



    mError("anode:%s, failed to create since %s", createReq.url, tstrerror(code));





×







391



  }










392













UNCOV

393



  mndReleaseAnode(pMnode, pObj);




×








UNCOV

394



  tFreeSMCreateAnodeReq(&createReq);




×








UNCOV

395



  TAOS_RETURN(code);




×







396



}










397












398



static int32_t mndUpdateAnode(SMnode *pMnode, SAnodeObj *pAnode, SRpcMsg *pReq) {




×







399



  mInfo("anode:%d, start to update", pAnode->id);





×







400



  int32_t   code = -1;




×







401



  STrans   *pTrans = NULL;




×







402



  SAnodeObj anodeObj = {0};




×







403



  anodeObj.id = pAnode->id;




×







404



  anodeObj.updateTime = taosGetTimestampMs();




×







405












406



  code = mndGetAnodeAlgoList(pAnode->url, &anodeObj);




×







407



  if (code != 0) goto _OVER;





×







408












409



  pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_CONFLICT_NOTHING, pReq, "update-anode");




×







410



  if (pTrans == NULL) {





×







411



    code = TSDB_CODE_MND_RETURN_VALUE_NULL;




×







412



    if (terrno != 0) code = terrno;





×







413



    goto _OVER;




×







414



  }










415



  mInfo("trans:%d, used to update anode:%d", pTrans->id, anodeObj.id);





×







416












417



  TAOS_CHECK_GOTO(mndSetCreateAnodeCommitLogs(pTrans, &anodeObj), NULL, _OVER);





×







418



  TAOS_CHECK_GOTO(mndTransPrepare(pMnode, pTrans), NULL, _OVER);





×







419



  code = 0;




×







420












421



_OVER:




×







422



  mndFreeAnode(&anodeObj);




×







423



  mndTransDrop(pTrans);




×







424



  TAOS_RETURN(code);




×







425



}










426












427



static int32_t mndUpdateAllAnodes(SMnode *pMnode, SRpcMsg *pReq) {




×







428



  mInfo("update all anodes");





×







429



  SSdb   *pSdb = pMnode->pSdb;




×







430



  int32_t code = 0;




×







431



  int32_t rows = 0;




×







432



  int32_t numOfRows = sdbGetSize(pSdb, SDB_ANODE);




×







433












434



  void *pIter = NULL;




×







435



  while (1) {




×







436



    SAnodeObj *pObj = NULL;




×







437



    ESdbStatus objStatus = 0;




×







438



    pIter = sdbFetchAll(pSdb, SDB_ANODE, pIter, (void **)&pObj, &objStatus, true);




×







439



    if (pIter == NULL) break;





×







440












441



    rows++;




×







442



    void *transReq = NULL;




×







443



    if (rows == numOfRows) transReq = pReq;





×







444



    code = mndUpdateAnode(pMnode, pObj, transReq);




×







445



    sdbRelease(pSdb, pObj);




×







446












447



    if (code != 0) break;





×







448



  }










449












450



  if (code == 0 && rows == numOfRows) {





×







451



    code = TSDB_CODE_ACTION_IN_PROGRESS;




×







452



  }










453












454



  return code;




×







455



}










456












457



static int32_t mndProcessUpdateAnodeReq(SRpcMsg *pReq) {




×







458



  SMnode          *pMnode = pReq->info.node;




×







459



  int32_t          code = -1;




×







460



  SAnodeObj       *pObj = NULL;




×







461



  SMUpdateAnodeReq updateReq = {0};




×







462












463



  TAOS_CHECK_GOTO(tDeserializeSMUpdateAnodeReq(pReq->pCont, pReq->contLen, &updateReq), NULL, _OVER);





×







464



  TAOS_CHECK_GOTO(mndCheckOperPrivilege(pMnode, pReq->info.conn.user, MND_OPER_UPDATE_ANODE), NULL, _OVER);





×







465












466



  if (updateReq.anodeId == -1) {





×







467



    code = mndUpdateAllAnodes(pMnode, pReq);




×







468



  } else {










469



    pObj = mndAcquireAnode(pMnode, updateReq.anodeId);




×







470



    if (pObj == NULL) {





×







471



      code = TSDB_CODE_MND_ANODE_NOT_EXIST;




×







472



      goto _OVER;




×







473



    }










474



    code = mndUpdateAnode(pMnode, pObj, pReq);




×







475



    if (code == 0) code = TSDB_CODE_ACTION_IN_PROGRESS;





×







476



  }










477












478



_OVER:




×







479



  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {





×







480



    if (updateReq.anodeId != -1) {





×







481



      mError("anode:%d, failed to update since %s", updateReq.anodeId, tstrerror(code));





×







482



    }










483



  }










484












485



  mndReleaseAnode(pMnode, pObj);




×







486



  tFreeSMUpdateAnodeReq(&updateReq);




×







487



  TAOS_RETURN(code);




×







488



}










489













UNCOV

490



static int32_t mndSetDropAnodeRedoLogs(STrans *pTrans, SAnodeObj *pObj) {




×








UNCOV

491



  int32_t  code = 0;




×








UNCOV

492



  SSdbRaw *pRedoRaw = mndAnodeActionEncode(pObj);




×








UNCOV

493



  if (pRedoRaw == NULL) {





×







494



    code = terrno;




×







495



    return code;




×







496



  }










497













UNCOV

498



  TAOS_CHECK_RETURN(mndTransAppendRedolog(pTrans, pRedoRaw));





×








UNCOV

499



  TAOS_CHECK_RETURN(sdbSetRawStatus(pRedoRaw, SDB_STATUS_DROPPING));





×







500













UNCOV

501



  return code;




×







502



}










503













UNCOV

504



static int32_t mndSetDropAnodeCommitLogs(STrans *pTrans, SAnodeObj *pObj) {




×








UNCOV

505



  int32_t  code = 0;




×








UNCOV

506



  SSdbRaw *pCommitRaw = mndAnodeActionEncode(pObj);




×








UNCOV

507



  if (pCommitRaw == NULL) {





×







508



    code = terrno;




×







509



    return code;




×







510



  }










511













UNCOV

512



  TAOS_CHECK_RETURN(mndTransAppendCommitlog(pTrans, pCommitRaw));





×








UNCOV

513



  TAOS_CHECK_RETURN(sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED));





×








UNCOV

514



  TAOS_RETURN(code);




×







515



}










516













UNCOV

517



static int32_t mndSetDropAnodeInfoToTrans(SMnode *pMnode, STrans *pTrans, SAnodeObj *pObj, bool force) {




×








UNCOV

518



  if (pObj == NULL) return 0;





×








UNCOV

519



  TAOS_CHECK_RETURN(mndSetDropAnodeRedoLogs(pTrans, pObj));





×








UNCOV

520



  TAOS_CHECK_RETURN(mndSetDropAnodeCommitLogs(pTrans, pObj));





×








UNCOV

521



  return 0;




×







522



}










523













UNCOV

524



static int32_t mndDropAnode(SMnode *pMnode, SRpcMsg *pReq, SAnodeObj *pObj) {




×








UNCOV

525



  int32_t code = 0;




×








UNCOV

526



  int32_t lino = 0;




×







527













UNCOV

528



  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_RETRY, TRN_CONFLICT_NOTHING, pReq, "drop-anode");




×








UNCOV

529



  TSDB_CHECK_NULL(pTrans, code, lino, _OVER, terrno);





×







530













UNCOV

531



  mndTransSetSerial(pTrans);




×








UNCOV

532



  mInfo("trans:%d, to drop anode:%d", pTrans->id, pObj->id);





×







533













UNCOV

534



  code = mndSetDropAnodeInfoToTrans(pMnode, pTrans, pObj, false);




×








UNCOV

535



  mndReleaseAnode(pMnode, pObj);




×







536













UNCOV

537



  TSDB_CHECK_CODE(code, lino, _OVER);





×







538













UNCOV

539



  code = mndTransPrepare(pMnode, pTrans);




×







540













UNCOV

541



_OVER:




×








UNCOV

542



  mndTransDrop(pTrans);




×








UNCOV

543



  return code;




×







544



}










545













UNCOV

546



static int32_t mndProcessDropAnodeReq(SRpcMsg *pReq) {




×








UNCOV

547



  SMnode        *pMnode = pReq->info.node;




×








UNCOV

548



  int32_t        code = -1;




×








UNCOV

549



  SAnodeObj     *pObj = NULL;




×








UNCOV

550



  SMDropAnodeReq dropReq = {0};




×







551













UNCOV

552



  TAOS_CHECK_GOTO(tDeserializeSMDropAnodeReq(pReq->pCont, pReq->contLen, &dropReq), NULL, _OVER);





×







553













UNCOV

554



  mInfo("anode:%d, start to drop", dropReq.anodeId);





×








UNCOV

555



  TAOS_CHECK_GOTO(mndCheckOperPrivilege(pMnode, pReq->info.conn.user, MND_OPER_DROP_ANODE), NULL, _OVER);





×







556













UNCOV

557



  if (dropReq.anodeId <= 0) {





×







558



    code = TSDB_CODE_INVALID_MSG;




×







559



    goto _OVER;




×







560



  }










561













UNCOV

562



  pObj = mndAcquireAnode(pMnode, dropReq.anodeId);




×








UNCOV

563



  if (pObj == NULL) {





×








UNCOV

564



    code = terrno;




×








UNCOV

565



    goto _OVER;




×







566



  }










567













UNCOV

568



  code = mndDropAnode(pMnode, pReq, pObj);




×








UNCOV

569



  if (code == 0) {





×








UNCOV

570



    code = TSDB_CODE_ACTION_IN_PROGRESS;




×







571



  }










572












573



_OVER:




×








UNCOV

574



  if (code != 0 && code != TSDB_CODE_ACTION_IN_PROGRESS) {





×








UNCOV

575



    mError("anode:%d, failed to drop since %s", dropReq.anodeId, tstrerror(code));





×







576



  }










577













UNCOV

578



  tFreeSMDropAnodeReq(&dropReq);




×








UNCOV

579



  TAOS_RETURN(code);




×







580



}










581













UNCOV

582



static int32_t mndRetrieveAnodes(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {




×








UNCOV

583



  SMnode    *pMnode = pReq->info.node;




×








UNCOV

584



  SSdb      *pSdb = pMnode->pSdb;




×








UNCOV

585



  int32_t    numOfRows = 0;




×








UNCOV

586



  int32_t    cols = 0;




×








UNCOV

587



  SAnodeObj *pObj = NULL;




×







588



  char       buf[TSDB_ANALYTIC_ANODE_URL_LEN + VARSTR_HEADER_SIZE];










589



  char       status[64];











UNCOV

590



  int32_t    code = 0;




×







591













UNCOV

592



  while (numOfRows < rows) {





×








UNCOV

593



    pShow->pIter = sdbFetch(pSdb, SDB_ANODE, pShow->pIter, (void **)&pObj);




×








UNCOV

594



    if (pShow->pIter == NULL) break;





×







595













UNCOV

596



    cols = 0;




×








UNCOV

597



    SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×








UNCOV

598



    code = colDataSetVal(pColInfo, numOfRows, (const char *)&pObj->id, false);




×








UNCOV

599



    if (code != 0) goto _end;





×







600













UNCOV

601



    STR_WITH_MAXSIZE_TO_VARSTR(buf, pObj->url, pShow->pMeta->pSchemas[cols].bytes);




×








UNCOV

602



    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×








UNCOV

603



    code = colDataSetVal(pColInfo, numOfRows, (const char *)buf, false);




×








UNCOV

604



    if (code != 0) goto _end;





×







605













UNCOV

606



    status[0] = 0;




×








UNCOV

607



    if (mndGetAnodeStatus(pObj, status, 64) == 0) {





×








UNCOV

608



      STR_TO_VARSTR(buf, status);




×







609



    } else {










610



      STR_TO_VARSTR(buf, "offline");




×







611



    }











UNCOV

612



    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×








UNCOV

613



    code = colDataSetVal(pColInfo, numOfRows, buf, false);




×








UNCOV

614



    if (code != 0) goto _end;





×







615













UNCOV

616



    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×








UNCOV

617



    code = colDataSetVal(pColInfo, numOfRows, (const char *)&pObj->createdTime, false);




×








UNCOV

618



    if (code != 0) goto _end;





×







619













UNCOV

620



    pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×








UNCOV

621



    code = colDataSetVal(pColInfo, numOfRows, (const char *)&pObj->updateTime, false);




×








UNCOV

622



    if (code != 0) goto _end;





×







623













UNCOV

624



    numOfRows++;




×








UNCOV

625



    sdbRelease(pSdb, pObj);




×







626



  }










627












628



_end:




×








UNCOV

629



  if (code != 0) sdbRelease(pSdb, pObj);





×







630













UNCOV

631



  pShow->numOfRows += numOfRows;




×








UNCOV

632



  return numOfRows;




×







633



}










634












635



static void mndCancelGetNextAnode(SMnode *pMnode, void *pIter) {




×







636



  SSdb *pSdb = pMnode->pSdb;




×







637



  sdbCancelFetchByType(pSdb, pIter, SDB_ANODE);




×







638



}




×







639












640



// todo handle multiple anode case, remove the duplicate algos











UNCOV

641



void mndRetrieveAlgoList(SMnode* pMnode, SArray* pFc, SArray* pAd) {




×








UNCOV

642



  SSdb      *pSdb = pMnode->pSdb;




×








UNCOV

643



  void      *pIter = NULL;




×








UNCOV

644



  SAnodeObj *pObj = NULL;




×







645












646



  while (1) {











UNCOV

647



    pIter = sdbFetch(pSdb, SDB_ANODE, pIter, (void **)&pObj);




×








UNCOV

648



    if (pIter == NULL) {





×








UNCOV

649



      break;




×







650



    }










651












652



    if (pObj->numOfAlgos >= ANAL_ALGO_TYPE_END) {





×







653



      if (pObj->algos[ANAL_ALGO_TYPE_ANOMALY_DETECT] != NULL) {





×







654



        taosArrayAddAll(pAd, pObj->algos[ANAL_ALGO_TYPE_ANOMALY_DETECT]);




×







655



      }










656












657



      if (pObj->algos[ANAL_ALGO_TYPE_FORECAST] != NULL) {





×







658



        taosArrayAddAll(pFc, pObj->algos[ANAL_ALGO_TYPE_FORECAST]);




×







659



      }










660



    }










661












662



    sdbRelease(pSdb, pObj);




×







663



  }











UNCOV

664



}




×







665













UNCOV

666



static int32_t mndRetrieveAnodesFull(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {




×








UNCOV

667



  SMnode    *pMnode = pReq->info.node;




×








UNCOV

668



  SSdb      *pSdb = pMnode->pSdb;




×








UNCOV

669



  int32_t    numOfRows = 0;




×








UNCOV

670



  int32_t    cols = 0;




×








UNCOV

671



  SAnodeObj *pObj = NULL;




×







672



  char       buf[TSDB_ANALYTIC_ALGO_NAME_LEN + VARSTR_HEADER_SIZE];











UNCOV

673



  int32_t    code = 0;




×







674













UNCOV

675



  while (numOfRows < rows) {





×








UNCOV

676



    pShow->pIter = sdbFetch(pSdb, SDB_ANODE, pShow->pIter, (void **)&pObj);




×








UNCOV

677



    if (pShow->pIter == NULL) break;





×







678













UNCOV

679



    for (int32_t t = 0; t < pObj->numOfAlgos; ++t) {





×








UNCOV

680



      SArray *algos = pObj->algos[t];




×







681













UNCOV

682



      for (int32_t a = 0; a < taosArrayGetSize(algos); ++a) {





×








UNCOV

683



        SAnodeAlgo *algo = taosArrayGet(algos, a);




×







684













UNCOV

685



        cols = 0;




×








UNCOV

686



        SColumnInfoData *pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×








UNCOV

687



        code = colDataSetVal(pColInfo, numOfRows, (const char *)&pObj->id, false);




×








UNCOV

688



        if (code != 0) goto _end;





×







689













UNCOV

690



        STR_TO_VARSTR(buf, taosAnalysisAlgoType(t));




×








UNCOV

691



        pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×








UNCOV

692



        code = colDataSetVal(pColInfo, numOfRows, buf, false);




×








UNCOV

693



        if (code != 0) goto _end;





×







694













UNCOV

695



        STR_TO_VARSTR(buf, algo->name);




×








UNCOV

696



        pColInfo = taosArrayGet(pBlock->pDataBlock, cols++);




×








UNCOV

697



        code = colDataSetVal(pColInfo, numOfRows, buf, false);




×








UNCOV

698



        if (code != 0) goto _end;





×







699













UNCOV

700



        numOfRows++;




×







701



      }










702



    }










703













UNCOV

704



    sdbRelease(pSdb, pObj);




×







705



  }










706












707



_end:




×








UNCOV

708



  if (code != 0) sdbRelease(pSdb, pObj);





×







709













UNCOV

710



  pShow->numOfRows += numOfRows;




×








UNCOV

711



  return numOfRows;




×







712



}










713












714



static void mndCancelGetNextAnodeFull(SMnode *pMnode, void *pIter) {




×







715



  SSdb *pSdb = pMnode->pSdb;




×







716



  sdbCancelFetchByType(pSdb, pIter, SDB_ANODE);




×







717



}




×







718













UNCOV

719



static int32_t mndDecodeAlgoList(SJson *pJson, SAnodeObj *pObj) {




×








UNCOV

720



  int32_t code = 0;




×








UNCOV

721



  int32_t protocol = 0;




×








UNCOV

722



  double  tmp = 0;




×








UNCOV

723



  char    buf[TSDB_ANALYTIC_ALGO_NAME_LEN + 1] = {0};




×







724













UNCOV

725



  code = tjsonGetDoubleValue(pJson, "protocol", &tmp);




×








UNCOV

726



  if (code < 0) return TSDB_CODE_INVALID_JSON_FORMAT;





×








UNCOV

727



  protocol = (int32_t)(tmp * 1000);




×








UNCOV

728



  if (protocol != 100 && protocol != 1000) return TSDB_CODE_MND_ANODE_INVALID_PROTOCOL;





×







729













UNCOV

730



  code = tjsonGetDoubleValue(pJson, "version", &tmp);




×








UNCOV

731



  pObj->version = (int32_t)(tmp * 1000);




×







732



#if 0










733



  if (code < 0) return TSDB_CODE_INVALID_JSON_FORMAT;










734



  if (pObj->version <= 0) return TSDB_CODE_MND_ANODE_INVALID_VERSION;










735



#endif










736













UNCOV

737



  SJson *details = tjsonGetObjectItem(pJson, "details");




×








UNCOV

738



  if (details == NULL) return TSDB_CODE_INVALID_JSON_FORMAT;





×








UNCOV

739



  int32_t numOfDetails = tjsonGetArraySize(details);




×







740













UNCOV

741



  pObj->algos = taosMemoryCalloc(ANAL_ALGO_TYPE_END, sizeof(SArray *));





×








UNCOV

742



  if (pObj->algos == NULL) return TSDB_CODE_OUT_OF_MEMORY;





×







743













UNCOV

744



  pObj->numOfAlgos = ANAL_ALGO_TYPE_END;




×








UNCOV

745



  for (int32_t i = 0; i < ANAL_ALGO_TYPE_END; ++i) {





×








UNCOV

746



    pObj->algos[i] = taosArrayInit(4, sizeof(SAnodeAlgo));




×








UNCOV

747



    if (pObj->algos[i] == NULL) return TSDB_CODE_OUT_OF_MEMORY;





×







748



  }










749













UNCOV

750



  for (int32_t d = 0; d < numOfDetails; ++d) {





×








UNCOV

751



    SJson *detail = tjsonGetArrayItem(details, d);




×








UNCOV

752



    if (detail == NULL) return TSDB_CODE_INVALID_JSON_FORMAT;





×







753













UNCOV

754



    code = tjsonGetStringValue2(detail, "type", buf, sizeof(buf));




×








UNCOV

755



    if (code < 0) return TSDB_CODE_INVALID_JSON_FORMAT;





×








UNCOV

756



    EAnalAlgoType type = taosAnalAlgoInt(buf);




×








UNCOV

757



    if (type < 0 || type >= ANAL_ALGO_TYPE_END) return TSDB_CODE_MND_ANODE_INVALID_ALGO_TYPE;





×







758













UNCOV

759



    SJson *algos = tjsonGetObjectItem(detail, "algo");




×








UNCOV

760



    if (algos == NULL) return TSDB_CODE_INVALID_JSON_FORMAT;





×








UNCOV

761



    int32_t numOfAlgos = tjsonGetArraySize(algos);




×








UNCOV

762



    for (int32_t a = 0; a < numOfAlgos; ++a) {





×








UNCOV

763



      SJson *algo = tjsonGetArrayItem(algos, a);




×








UNCOV

764



      if (algo == NULL) return TSDB_CODE_INVALID_JSON_FORMAT;





×







765













UNCOV

766



      code = tjsonGetStringValue2(algo, "name", buf, sizeof(buf));




×








UNCOV

767



      if (code < 0) return TSDB_CODE_MND_ANODE_TOO_LONG_ALGO_NAME;





×







768













UNCOV

769



      SAnodeAlgo algoObj = {0};




×








UNCOV

770



      algoObj.nameLen = strlen(buf) + 1;




×








UNCOV

771



      if (algoObj.nameLen <= 1) return TSDB_CODE_INVALID_JSON_FORMAT;





×








UNCOV

772



      algoObj.name = taosMemoryCalloc(algoObj.nameLen, 1);





×








UNCOV

773



      tstrncpy(algoObj.name, buf, algoObj.nameLen);




×







774













UNCOV

775



      if (taosArrayPush(pObj->algos[type], &algoObj) == NULL) return TSDB_CODE_OUT_OF_MEMORY;





×







776



    }










777



  }










778













UNCOV

779



  return 0;




×







780



}










781













UNCOV

782



static int32_t mndGetAnodeAlgoList(const char *url, SAnodeObj *pObj) {




×








UNCOV

783



  char anodeUrl[TSDB_ANALYTIC_ANODE_URL_LEN + 1] = {0};




×








UNCOV

784



  snprintf(anodeUrl, TSDB_ANALYTIC_ANODE_URL_LEN, "%s/%s", url, "list");




×







785













UNCOV

786



  SJson *pJson = taosAnalSendReqRetJson(anodeUrl, ANALYTICS_HTTP_TYPE_GET, NULL);




×








UNCOV

787



  if (pJson == NULL) return terrno;





×







788













UNCOV

789



  int32_t code = mndDecodeAlgoList(pJson, pObj);




×








UNCOV

790



  if (pJson != NULL) tjsonDelete(pJson);





×







791













UNCOV

792



  TAOS_RETURN(code);




×







793



}










794













UNCOV

795



static int32_t mndGetAnodeStatus(SAnodeObj *pObj, char *status, int32_t statusLen) {




×








UNCOV

796



  int32_t code = 0;




×








UNCOV

797



  int32_t protocol = 0;




×








UNCOV

798



  double  tmp = 0;




×








UNCOV

799



  char    anodeUrl[TSDB_ANALYTIC_ANODE_URL_LEN + 1] = {0};




×








UNCOV

800



  snprintf(anodeUrl, TSDB_ANALYTIC_ANODE_URL_LEN, "%s/%s", pObj->url, "status");




×







801













UNCOV

802



  SJson *pJson = taosAnalSendReqRetJson(anodeUrl, ANALYTICS_HTTP_TYPE_GET, NULL);




×








UNCOV

803



  if (pJson == NULL) return terrno;





×







804













UNCOV

805



  code = tjsonGetDoubleValue(pJson, "protocol", &tmp);




×








UNCOV

806



  if (code < 0) {





×







807



    code = TSDB_CODE_INVALID_JSON_FORMAT;




×







808



    goto _OVER;




×







809



  }











UNCOV

810



  protocol = (int32_t)(tmp * 1000);




×








UNCOV

811



  if (protocol != 100 && protocol != 1000) {





×







812



    code = TSDB_CODE_MND_ANODE_INVALID_PROTOCOL;




×







813



    goto _OVER;




×







814



  }










815













UNCOV

816



  code = tjsonGetStringValue2(pJson, "status", status, statusLen);




×








UNCOV

817



  if (code < 0) {





×







818



    code = TSDB_CODE_INVALID_JSON_FORMAT;




×







819



    goto _OVER;




×







820



  }











UNCOV

821



  if (strlen(status) == 0) {





×







822



    code = TSDB_CODE_MND_ANODE_INVALID_PROTOCOL;




×







823



    goto _OVER;




×







824



  }










825













UNCOV

826



_OVER:




×








UNCOV

827



  if (pJson != NULL) tjsonDelete(pJson);





×








UNCOV

828



  TAOS_RETURN(code);




×







829



}










830













UNCOV

831



static int32_t mndProcessAnalAlgoReq(SRpcMsg *pReq) {




×








UNCOV

832



  SMnode              *pMnode = pReq->info.node;




×








UNCOV

833



  SSdb                *pSdb = pMnode->pSdb;




×








UNCOV

834



  int32_t              code = -1;




×








UNCOV

835



  SAnodeObj           *pObj = NULL;




×







836



  SAnalyticsUrl             url;










837



  int32_t              nameLen;










838



  char                 name[TSDB_ANALYTIC_ALGO_KEY_LEN];











UNCOV

839



  SRetrieveAnalAlgoReq req = {0};




×








UNCOV

840



  SRetrieveAnalAlgoRsp rsp = {0};




×







841













UNCOV

842



  TAOS_CHECK_GOTO(tDeserializeRetrieveAnalAlgoReq(pReq->pCont, pReq->contLen, &req), NULL, _OVER);





×







843













UNCOV

844



  rsp.ver = sdbGetTableVer(pSdb, SDB_ANODE);




×








UNCOV

845



  if (req.analVer != rsp.ver) {





×








UNCOV

846



    mInfo("dnode:%d, update analysis old ver:%" PRId64 " to new ver:%" PRId64, req.dnodeId, req.analVer, rsp.ver);





×








UNCOV

847



    rsp.hash = taosHashInit(64, MurmurHash3_32, true, HASH_ENTRY_LOCK);




×








UNCOV

848



    if (rsp.hash == NULL) {





×







849



      terrno = TSDB_CODE_OUT_OF_MEMORY;




×







850



      goto _OVER;




×







851



    }










852













UNCOV

853



    void *pIter = NULL;




×








UNCOV

854



    while (1) {




×








UNCOV

855



      SAnodeObj *pAnode = NULL;




×








UNCOV

856



      pIter = sdbFetch(pSdb, SDB_ANODE, pIter, (void **)&pAnode);




×








UNCOV

857



      if (pIter == NULL) break;





×







858













UNCOV

859



      url.anode = pAnode->id;




×








UNCOV

860



      for (int32_t t = 0; t < pAnode->numOfAlgos; ++t) {





×








UNCOV

861



        SArray *algos = pAnode->algos[t];




×








UNCOV

862



        url.type = t;




×







863













UNCOV

864



        for (int32_t a = 0; a < taosArrayGetSize(algos); ++a) {





×








UNCOV

865



          SAnodeAlgo *algo = taosArrayGet(algos, a);




×








UNCOV

866



          nameLen = 1 + tsnprintf(name, sizeof(name) - 1, "%d:%s", url.type, algo->name);




×







867













UNCOV

868



          SAnalyticsUrl *pOldUrl = taosHashAcquire(rsp.hash, name, nameLen);




×








UNCOV

869



          if (pOldUrl == NULL || (pOldUrl != NULL && pOldUrl->anode < url.anode)) {





×








UNCOV

870



            if (pOldUrl != NULL) {





×







871



              taosMemoryFreeClear(pOldUrl->url);





×







872



              if (taosHashRemove(rsp.hash, name, nameLen) != 0) {





×







873



                sdbRelease(pSdb, pAnode);




×







874



                goto _OVER;




×







875



              }










876



            }











UNCOV

877



            url.url = taosMemoryMalloc(TSDB_ANALYTIC_ANODE_URL_LEN + TSDB_ANALYTIC_ALGO_TYPE_LEN + 1);





×








UNCOV

878



            if (url.url == NULL) {





×







879



              sdbRelease(pSdb, pAnode);




×







880



              goto _OVER;




×







881



            }










882













UNCOV

883



            url.urlLen = 1 + tsnprintf(url.url, TSDB_ANALYTIC_ANODE_URL_LEN + TSDB_ANALYTIC_ALGO_TYPE_LEN, "%s/%s", pAnode->url,




×







884



                                      taosAnalAlgoUrlStr(url.type));











UNCOV

885



            if (taosHashPut(rsp.hash, name, nameLen, &url, sizeof(SAnalyticsUrl)) != 0) {





×







886



              taosMemoryFree(url.url);





×







887



              sdbRelease(pSdb, pAnode);




×







888



              goto _OVER;




×







889



            }










890



          }










891



        }










892













UNCOV

893



        sdbRelease(pSdb, pAnode);




×







894



      }










895



    }










896



  }










897













UNCOV

898



  int32_t contLen = tSerializeRetrieveAnalAlgoRsp(NULL, 0, &rsp);




×








UNCOV

899



  void   *pHead = rpcMallocCont(contLen);




×








UNCOV

900



  (void)tSerializeRetrieveAnalAlgoRsp(pHead, contLen, &rsp);




×







901













UNCOV

902



  pReq->info.rspLen = contLen;




×








UNCOV

903



  pReq->info.rsp = pHead;




×







904













UNCOV

905



_OVER:




×








UNCOV

906



  tFreeRetrieveAnalAlgoRsp(&rsp);




×








UNCOV

907



  TAOS_RETURN(code);




×







908



}










909












910



#else










911












912



static int32_t mndProcessUnsupportReq(SRpcMsg *pReq) { return TSDB_CODE_OPS_NOT_SUPPORT; }










913



static int32_t mndRetrieveUnsupport(SRpcMsg *pReq, SShowObj *pShow, SSDataBlock *pBlock, int32_t rows) {










914



  return TSDB_CODE_OPS_NOT_SUPPORT;










915



}










916












917



int32_t mndInitAnode(SMnode *pMnode) {










918



  mndSetMsgHandle(pMnode, TDMT_MND_CREATE_ANODE, mndProcessUnsupportReq);










919



  mndSetMsgHandle(pMnode, TDMT_MND_UPDATE_ANODE, mndProcessUnsupportReq);










920



  mndSetMsgHandle(pMnode, TDMT_MND_DROP_ANODE, mndProcessUnsupportReq);










921



  mndSetMsgHandle(pMnode, TDMT_MND_RETRIEVE_ANAL_ALGO, mndProcessUnsupportReq);










922












923



  mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_ANODE, mndRetrieveUnsupport);










924



  mndAddShowRetrieveHandle(pMnode, TSDB_MGMT_TABLE_ANODE_FULL, mndRetrieveUnsupport);










925



  return 0;










926



}










927












928



void mndCleanupAnode(SMnode *pMnode) {}










929



void mndRetrieveAlgoList(SMnode *pMnode, SArray *pFc, SArray *pAd) {}










930












931



#endif























    

  • 

    •