mnodeMnode.c

/*
 * 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/>.
 */

#define _DEFAULT_SOURCE
#include "os.h"
#include "taoserror.h"
#include "tglobal.h"
#include "trpc.h"
#include "tsync.h"
#include "tbalance.h"
#include "tutil.h"
#include "tsocket.h"
#include "tdataformat.h"
#include "dnode.h"
#include "mnode.h"
#include "mnodeDef.h"
#include "mnodeInt.h"
#include "mnodeMnode.h"
#include "mnodeDnode.h"
#include "mnodeSdb.h"
#include "mnodeShow.h"
#include "mnodeUser.h"
#include "mnodeVgroup.h"

static void *        tsMnodeSdb = NULL;
static int32_t       tsMnodeUpdateSize = 0;
static SRpcEpSet     tsMnodeEpSetForShell;
static SRpcEpSet     tsMnodeEpSetForPeer;
static SDMMnodeInfos tsMnodeInfos;
static int32_t mnodeGetMnodeMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn);
static int32_t mnodeRetrieveMnodes(SShowObj *pShow, char *data, int32_t rows, void *pConn);

#if defined(LINUX)
  static pthread_rwlock_t         tsMnodeLock;
  #define mnodeMnodeWrLock()      pthread_rwlock_wrlock(&tsMnodeLock)
  #define mnodeMnodeRdLock()      pthread_rwlock_rdlock(&tsMnodeLock)
  #define mnodeMnodeUnLock()      pthread_rwlock_unlock(&tsMnodeLock)
  #define mnodeMnodeInitLock()    pthread_rwlock_init(&tsMnodeLock, NULL)
  #define mnodeMnodeDestroyLock() pthread_rwlock_destroy(&tsMnodeLock)
#else
  static pthread_mutex_t          tsMnodeLock;
  #define mnodeMnodeWrLock()      pthread_mutex_lock(&tsMnodeLock)
  #define mnodeMnodeRdLock()      pthread_mutex_lock(&tsMnodeLock)
  #define mnodeMnodeUnLock()      pthread_mutex_unlock(&tsMnodeLock)
  #define mnodeMnodeInitLock()    pthread_mutex_init(&tsMnodeLock, NULL)
  #define mnodeMnodeDestroyLock() pthread_mutex_destroy(&tsMnodeLock)
#endif

static int32_t mnodeMnodeActionDestroy(SSdbOper *pOper) {
  taosTFree(pOper->pObj);
  return TSDB_CODE_SUCCESS;
}

static int32_t mnodeMnodeActionInsert(SSdbOper *pOper) {
  SMnodeObj *pMnode = pOper->pObj;
  SDnodeObj *pDnode = mnodeGetDnode(pMnode->mnodeId);
  if (pDnode == NULL) return TSDB_CODE_MND_DNODE_NOT_EXIST;

  pDnode->isMgmt = true;
  mnodeDecDnodeRef(pDnode);
  
  mInfo("mnode:%d, fqdn:%s ep:%s port:%d, do insert action", pMnode->mnodeId, pDnode->dnodeFqdn, pDnode->dnodeEp, pDnode->dnodePort);
  return TSDB_CODE_SUCCESS;
}

static int32_t mnodeMnodeActionDelete(SSdbOper *pOper) {
  SMnodeObj *pMnode = pOper->pObj;

  SDnodeObj *pDnode = mnodeGetDnode(pMnode->mnodeId);
  if (pDnode == NULL) return TSDB_CODE_MND_DNODE_NOT_EXIST;
  pDnode->isMgmt = false;
  mnodeDecDnodeRef(pDnode);

  mDebug("mnode:%d, is dropped from sdb", pMnode->mnodeId);
  return TSDB_CODE_SUCCESS;
}

static int32_t mnodeMnodeActionUpdate(SSdbOper *pOper) {
  SMnodeObj *pMnode = pOper->pObj;
  SMnodeObj *pSaved = mnodeGetMnode(pMnode->mnodeId);
  if (pMnode != pSaved) {
    memcpy(pSaved, pMnode, pOper->rowSize);
    free(pMnode);
  }
  mnodeDecMnodeRef(pSaved);
  return TSDB_CODE_SUCCESS;
}

static int32_t mnodeMnodeActionEncode(SSdbOper *pOper) {
  SMnodeObj *pMnode = pOper->pObj;
  memcpy(pOper->rowData, pMnode, tsMnodeUpdateSize);
  pOper->rowSize = tsMnodeUpdateSize;
  return TSDB_CODE_SUCCESS;
}

static int32_t mnodeMnodeActionDecode(SSdbOper *pOper) {
  SMnodeObj *pMnode = calloc(1, sizeof(SMnodeObj));
  if (pMnode == NULL) return TSDB_CODE_MND_OUT_OF_MEMORY;

  memcpy(pMnode, pOper->rowData, tsMnodeUpdateSize);
  pOper->pObj = pMnode;
  return TSDB_CODE_SUCCESS;
}

static int32_t mnodeMnodeActionRestored() {
  if (mnodeGetMnodesNum() == 1) {
    SMnodeObj *pMnode = NULL;
    void *pIter = mnodeGetNextMnode(NULL, &pMnode);
    if (pMnode != NULL) {
      pMnode->role = TAOS_SYNC_ROLE_MASTER;
      mnodeDecMnodeRef(pMnode);
    }
    sdbFreeIter(pIter);
  }

  mnodeUpdateMnodeEpSet();

  return TSDB_CODE_SUCCESS;
}

int32_t mnodeInitMnodes() {
  mnodeMnodeInitLock();

  SMnodeObj tObj;
  tsMnodeUpdateSize = (int8_t *)tObj.updateEnd - (int8_t *)&tObj;

  SSdbTableDesc tableDesc = {
    .tableId      = SDB_TABLE_MNODE,
    .tableName    = "mnodes",
    .hashSessions = TSDB_DEFAULT_MNODES_HASH_SIZE,
    .maxRowSize   = tsMnodeUpdateSize,
    .refCountPos  = (int8_t *)(&tObj.refCount) - (int8_t *)&tObj,
    .keyType      = SDB_KEY_INT,
    .insertFp     = mnodeMnodeActionInsert,
    .deleteFp     = mnodeMnodeActionDelete,
    .updateFp     = mnodeMnodeActionUpdate,
    .encodeFp     = mnodeMnodeActionEncode,
    .decodeFp     = mnodeMnodeActionDecode,
    .destroyFp    = mnodeMnodeActionDestroy,
    .restoredFp   = mnodeMnodeActionRestored
  };

  tsMnodeSdb = sdbOpenTable(&tableDesc);
  if (tsMnodeSdb == NULL) {
    mError("failed to init mnodes data");
    return -1;
  }

  mnodeAddShowMetaHandle(TSDB_MGMT_TABLE_MNODE, mnodeGetMnodeMeta);
  mnodeAddShowRetrieveHandle(TSDB_MGMT_TABLE_MNODE, mnodeRetrieveMnodes);

  mDebug("table:mnodes table is created");
  return TSDB_CODE_SUCCESS;
}

void mnodeCleanupMnodes() {
  sdbCloseTable(tsMnodeSdb);
  tsMnodeSdb = NULL;
  mnodeMnodeDestroyLock();
}

int32_t mnodeGetMnodesNum() { 
  return sdbGetNumOfRows(tsMnodeSdb); 
}

void *mnodeGetMnode(int32_t mnodeId) {
  return sdbGetRow(tsMnodeSdb, &mnodeId);
}

void mnodeIncMnodeRef(SMnodeObj *pMnode) {
  sdbIncRef(tsMnodeSdb, pMnode);
}

void mnodeDecMnodeRef(SMnodeObj *pMnode) {
  sdbDecRef(tsMnodeSdb, pMnode);
}

void *mnodeGetNextMnode(void *pIter, SMnodeObj **pMnode) { 
  return sdbFetchRow(tsMnodeSdb, pIter, (void **)pMnode); 
}

char *mnodeGetMnodeRoleStr(int32_t role) {
  return syncRole[role];
}

void mnodeUpdateMnodeEpSet() {
  mInfo("update mnodes epSet, numOfEps:%d ", mnodeGetMnodesNum());

  mnodeMnodeWrLock();

  memset(&tsMnodeEpSetForShell, 0, sizeof(SRpcEpSet));
  memset(&tsMnodeEpSetForPeer, 0, sizeof(SRpcEpSet));
  memset(&tsMnodeInfos, 0, sizeof(SDMMnodeInfos));

  int32_t index = 0;
  void *  pIter = NULL;
  while (1) {
    SMnodeObj *pMnode = NULL;
    pIter = mnodeGetNextMnode(pIter, &pMnode);
    if (pMnode == NULL) break;

    SDnodeObj *pDnode = mnodeGetDnode(pMnode->mnodeId);
    if (pDnode != NULL) {
      strcpy(tsMnodeEpSetForShell.fqdn[index], pDnode->dnodeFqdn);
      tsMnodeEpSetForShell.port[index] = htons(pDnode->dnodePort);
      mDebug("mnode:%d, for shell fqdn:%s %d", pDnode->dnodeId, tsMnodeEpSetForShell.fqdn[index], htons(tsMnodeEpSetForShell.port[index]));      

      strcpy(tsMnodeEpSetForPeer.fqdn[index], pDnode->dnodeFqdn);
      tsMnodeEpSetForPeer.port[index] = htons(pDnode->dnodePort + TSDB_PORT_DNODEDNODE);
      mDebug("mnode:%d, for peer fqdn:%s %d", pDnode->dnodeId, tsMnodeEpSetForPeer.fqdn[index], htons(tsMnodeEpSetForPeer.port[index]));

      tsMnodeInfos.nodeInfos[index].nodeId = htonl(pMnode->mnodeId);
      strcpy(tsMnodeInfos.nodeInfos[index].nodeEp, pDnode->dnodeEp);

      if (pMnode->role == TAOS_SYNC_ROLE_MASTER) {
        tsMnodeEpSetForShell.inUse = index;
        tsMnodeEpSetForPeer.inUse = index;
        tsMnodeInfos.inUse = index;
      }

      mInfo("mnode:%d, ep:%s %s", pDnode->dnodeId, pDnode->dnodeEp, pMnode->role == TAOS_SYNC_ROLE_MASTER ? "master" : "");
      index++;
    }

    mnodeDecDnodeRef(pDnode);
    mnodeDecMnodeRef(pMnode);
  }

  tsMnodeInfos.nodeNum = index;
  tsMnodeEpSetForShell.numOfEps = index;
  tsMnodeEpSetForPeer.numOfEps = index;

  sdbFreeIter(pIter);

  mnodeMnodeUnLock();
}

void mnodeGetMnodeEpSetForPeer(SRpcEpSet *epSet) {
  mnodeMnodeRdLock();
  *epSet = tsMnodeEpSetForPeer;
  mnodeMnodeUnLock();
}

void mnodeGetMnodeEpSetForShell(SRpcEpSet *epSet) {
  mnodeMnodeRdLock();
  *epSet = tsMnodeEpSetForShell;
  mnodeMnodeUnLock();
}

char* mnodeGetMnodeMasterEp() {
  return tsMnodeInfos.nodeInfos[tsMnodeInfos.inUse].nodeEp;
}

void mnodeGetMnodeInfos(void *mnodeInfos) {
  mnodeMnodeRdLock();
  *(SDMMnodeInfos *)mnodeInfos = tsMnodeInfos;
  mnodeMnodeUnLock();
}

static int32_t mnodeSendCreateMnodeMsg(int32_t dnodeId, char *dnodeEp) {
  mDebug("dnode:%d, send create mnode msg to dnode %s", dnodeId, dnodeEp);

  SMDCreateMnodeMsg *pCreate = rpcMallocCont(sizeof(SMDCreateMnodeMsg));
  if (pCreate == NULL) {
    return TSDB_CODE_MND_OUT_OF_MEMORY;
  } else {
    pCreate->dnodeId = htonl(dnodeId);
    tstrncpy(pCreate->dnodeEp, dnodeEp, sizeof(pCreate->dnodeEp));
    pCreate->mnodes = tsMnodeInfos;
    bool found = false;
    for (int i = 0; i < pCreate->mnodes.nodeNum; ++i) {
      if (pCreate->mnodes.nodeInfos[i].nodeId == htonl(dnodeId)) {
        found = true;
      }
    }
    if (!found) {
      pCreate->mnodes.nodeInfos[pCreate->mnodes.nodeNum].nodeId = htonl(dnodeId);
      tstrncpy(pCreate->mnodes.nodeInfos[pCreate->mnodes.nodeNum].nodeEp, dnodeEp, sizeof(pCreate->dnodeEp));
      pCreate->mnodes.nodeNum++;
    }
  }

  SRpcMsg rpcMsg = {0};
  rpcMsg.pCont = pCreate;
  rpcMsg.contLen = sizeof(SMDCreateMnodeMsg);
  rpcMsg.msgType = TSDB_MSG_TYPE_MD_CREATE_MNODE;

  SRpcMsg   rpcRsp = {0};
  SRpcEpSet epSet = mnodeGetEpSetFromIp(pCreate->dnodeEp);
  dnodeSendMsgToDnodeRecv(&rpcMsg, &rpcRsp, &epSet);

  if (rpcRsp.code != TSDB_CODE_SUCCESS) {
    mError("dnode:%d, failed to send create mnode msg, ep:%s reason:%s", dnodeId, dnodeEp, tstrerror(rpcRsp.code));
  } else {
    mDebug("dnode:%d, create mnode msg is disposed, mnode is created in dnode", dnodeId);
  }

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

static int32_t mnodeCreateMnodeCb(SMnodeMsg *pMsg, int32_t code) {
  if (code != TSDB_CODE_SUCCESS) {
    mError("failed to create mnode, reason:%s", tstrerror(code));
  } else {
    mDebug("mnode is created successfully");
    mnodeUpdateMnodeEpSet();
    sdbUpdateSync(NULL);
  }

  return code;
}

void mnodeCreateMnode(int32_t dnodeId, char *dnodeEp, bool needConfirm) {
  SMnodeObj *pMnode = calloc(1, sizeof(SMnodeObj));
  pMnode->mnodeId = dnodeId;
  pMnode->createdTime = taosGetTimestampMs();

  SSdbOper oper = {
    .type  = SDB_OPER_GLOBAL,
    .table = tsMnodeSdb,
    .pObj  = pMnode,
    .writeCb = mnodeCreateMnodeCb
  };

  int32_t code = TSDB_CODE_SUCCESS;
  if (needConfirm) {
    code = mnodeSendCreateMnodeMsg(dnodeId, dnodeEp);
  }

  if (code != TSDB_CODE_SUCCESS) {
    taosTFree(pMnode);
    return;
  }

  code = sdbInsertRow(&oper);
  if (code != TSDB_CODE_SUCCESS && code != TSDB_CODE_MND_ACTION_IN_PROGRESS) {
    mError("dnode:%d, failed to create mnode, ep:%s reason:%s", dnodeId, dnodeEp, tstrerror(code));
    taosTFree(pMnode);
  }
}

void mnodeDropMnodeLocal(int32_t dnodeId) {
  SMnodeObj *pMnode = mnodeGetMnode(dnodeId);
  if (pMnode != NULL) {
    SSdbOper oper = {.type = SDB_OPER_LOCAL, .table = tsMnodeSdb, .pObj = pMnode};
    sdbDeleteRow(&oper);
    mnodeDecMnodeRef(pMnode);
  }

  mnodeUpdateMnodeEpSet();
  sdbUpdateSync(NULL);
}

int32_t mnodeDropMnode(int32_t dnodeId) {
  SMnodeObj *pMnode = mnodeGetMnode(dnodeId);
  if (pMnode == NULL) {
    return TSDB_CODE_MND_DNODE_NOT_EXIST;
  }
  
  SSdbOper oper = {
    .type = SDB_OPER_GLOBAL,
    .table = tsMnodeSdb,
    .pObj = pMnode
  };

  int32_t code = sdbDeleteRow(&oper);

  sdbDecRef(tsMnodeSdb, pMnode);

  mnodeUpdateMnodeEpSet();
  sdbUpdateSync(NULL);

  return code;
}

static int32_t mnodeGetMnodeMeta(STableMetaMsg *pMeta, SShowObj *pShow, void *pConn) {
  sdbUpdateMnodeRoles();
  SUserObj *pUser = mnodeGetUserFromConn(pConn);
  if (pUser == NULL) return 0;

  if (strcmp(pUser->pAcct->user, TSDB_DEFAULT_USER) != 0)  {
    mnodeDecUserRef(pUser);
    return TSDB_CODE_MND_NO_RIGHTS;
  }

  int32_t  cols = 0;
  SSchema *pSchema = pMeta->schema;

  pShow->bytes[cols] = 2;
  pSchema[cols].type = TSDB_DATA_TYPE_SMALLINT;
  strcpy(pSchema[cols].name, "id");
  pSchema[cols].bytes = htons(pShow->bytes[cols]);
  cols++;

  pShow->bytes[cols] = 40 + VARSTR_HEADER_SIZE;
  pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
  strcpy(pSchema[cols].name, "end_point");
  pSchema[cols].bytes = htons(pShow->bytes[cols]);
  cols++;

  pShow->bytes[cols] = 12 + VARSTR_HEADER_SIZE;
  pSchema[cols].type = TSDB_DATA_TYPE_BINARY;
  strcpy(pSchema[cols].name, "role");
  pSchema[cols].bytes = htons(pShow->bytes[cols]);
  cols++;
  
  pShow->bytes[cols] = 8;
  pSchema[cols].type = TSDB_DATA_TYPE_TIMESTAMP;
  strcpy(pSchema[cols].name, "create_time");
  pSchema[cols].bytes = htons(pShow->bytes[cols]);
  cols++;

  pMeta->numOfColumns = htons(cols);
  pShow->numOfColumns = cols;

  pShow->offset[0] = 0;
  for (int32_t i = 1; i < cols; ++i) {
    pShow->offset[i] = pShow->offset[i - 1] + pShow->bytes[i - 1];
  }

  pShow->numOfRows = mnodeGetMnodesNum();
  pShow->rowSize = pShow->offset[cols - 1] + pShow->bytes[cols - 1];
  pShow->pIter = NULL;
  mnodeDecUserRef(pUser);

  return 0;
}

static int32_t mnodeRetrieveMnodes(SShowObj *pShow, char *data, int32_t rows, void *pConn) {
  int32_t    numOfRows = 0;
  int32_t    cols      = 0;
  SMnodeObj *pMnode   = NULL;
  char      *pWrite;

  while (numOfRows < rows) {
    pShow->pIter = mnodeGetNextMnode(pShow->pIter, &pMnode);
    if (pMnode == NULL) break;

    cols = 0;

    pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
    *(int16_t *)pWrite = pMnode->mnodeId;
    cols++;

    pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
    
    SDnodeObj *pDnode = mnodeGetDnode(pMnode->mnodeId);
    if (pDnode != NULL) {
      STR_WITH_MAXSIZE_TO_VARSTR(pWrite, pDnode->dnodeEp, pShow->bytes[cols]);
    } else {
      STR_WITH_MAXSIZE_TO_VARSTR(pWrite, "invalid ep", pShow->bytes[cols]);
    }
    mnodeDecDnodeRef(pDnode);

    cols++;

    pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
    char* roles = mnodeGetMnodeRoleStr(pMnode->role);
    STR_WITH_MAXSIZE_TO_VARSTR(pWrite, roles, pShow->bytes[cols]);
    cols++;

    pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
    *(int64_t *)pWrite = pMnode->createdTime;
    cols++;
    
    numOfRows++;

    mnodeDecMnodeRef(pMnode);
  }
  mnodeVacuumResult(data, cols, numOfRows, rows, pShow);

  pShow->numOfReads += numOfRows;

  return numOfRows;
}