mndOffset.c 9.5 KB
Newer Older
L
Liu Jicong 已提交
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34
/*
 * 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 "mndOffset.h"
#include "mndAuth.h"
#include "mndDb.h"
#include "mndDnode.h"
#include "mndMnode.h"
#include "mndShow.h"
#include "mndStb.h"
#include "mndTrans.h"
#include "mndUser.h"
#include "mndVgroup.h"
#include "tname.h"

#define MND_OFFSET_VER_NUMBER   1
#define MND_OFFSET_RESERVE_SIZE 64

static int32_t mndOffsetActionInsert(SSdb *pSdb, SMqOffsetObj *pOffset);
static int32_t mndOffsetActionDelete(SSdb *pSdb, SMqOffsetObj *pOffset);
static int32_t mndOffsetActionUpdate(SSdb *pSdb, SMqOffsetObj *pOffset, SMqOffsetObj *pNewOffset);
S
Shengliang Guan 已提交
35
static int32_t mndProcessCommitOffsetReq(SRpcMsg *pReq);
L
Liu Jicong 已提交
36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52

int32_t mndInitOffset(SMnode *pMnode) {
  SSdbTable table = {.sdbType = SDB_OFFSET,
                     .keyType = SDB_KEY_BINARY,
                     .encodeFp = (SdbEncodeFp)mndOffsetActionEncode,
                     .decodeFp = (SdbDecodeFp)mndOffsetActionDecode,
                     .insertFp = (SdbInsertFp)mndOffsetActionInsert,
                     .updateFp = (SdbUpdateFp)mndOffsetActionUpdate,
                     .deleteFp = (SdbDeleteFp)mndOffsetActionDelete};

  mndSetMsgHandle(pMnode, TDMT_MND_MQ_COMMIT_OFFSET, mndProcessCommitOffsetReq);

  return sdbSetTable(pMnode->pSdb, table);
}

void mndCleanupOffset(SMnode *pMnode) {}

L
Liu Jicong 已提交
53 54 55 56 57 58 59 60
bool mndOffsetFromTopic(SMqOffsetObj *pOffset, const char *topic) {
  int32_t i = 0;
  while (pOffset->key[i] != ':') i++;
  while (pOffset->key[i] != ':') i++;
  if (strcmp(&pOffset->key[i + 1], topic) == 0) return true;
  return false;
}

L
Liu Jicong 已提交
61 62 63 64 65 66 67 68 69
SSdbRaw *mndOffsetActionEncode(SMqOffsetObj *pOffset) {
  terrno = TSDB_CODE_OUT_OF_MEMORY;
  void   *buf = NULL;
  int32_t tlen = tEncodeSMqOffsetObj(NULL, pOffset);
  int32_t size = sizeof(int32_t) + tlen + MND_OFFSET_RESERVE_SIZE;

  SSdbRaw *pRaw = sdbAllocRaw(SDB_OFFSET, MND_OFFSET_VER_NUMBER, size);
  if (pRaw == NULL) goto OFFSET_ENCODE_OVER;

wafwerar's avatar
wafwerar 已提交
70
  buf = taosMemoryMalloc(tlen);
L
Liu Jicong 已提交
71 72 73 74 75 76 77 78 79 80 81 82 83 84
  if (buf == NULL) goto OFFSET_ENCODE_OVER;

  void *abuf = buf;
  tEncodeSMqOffsetObj(&abuf, pOffset);

  int32_t dataPos = 0;
  SDB_SET_INT32(pRaw, dataPos, tlen, OFFSET_ENCODE_OVER);
  SDB_SET_BINARY(pRaw, dataPos, buf, tlen, OFFSET_ENCODE_OVER);
  SDB_SET_RESERVE(pRaw, dataPos, MND_OFFSET_RESERVE_SIZE, OFFSET_ENCODE_OVER);
  SDB_SET_DATALEN(pRaw, dataPos, OFFSET_ENCODE_OVER);

  terrno = TSDB_CODE_SUCCESS;

OFFSET_ENCODE_OVER:
wafwerar's avatar
wafwerar 已提交
85
  taosMemoryFreeClear(buf);
L
Liu Jicong 已提交
86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117
  if (terrno != TSDB_CODE_SUCCESS) {
    mError("offset:%s, failed to encode to raw:%p since %s", pOffset->key, pRaw, terrstr());
    sdbFreeRaw(pRaw);
    return NULL;
  }

  mTrace("offset:%s, encode to raw:%p, row:%p", pOffset->key, pRaw, pOffset);
  return pRaw;
}

SSdbRow *mndOffsetActionDecode(SSdbRaw *pRaw) {
  terrno = TSDB_CODE_OUT_OF_MEMORY;
  void *buf = NULL;

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

  if (sver != MND_OFFSET_VER_NUMBER) {
    terrno = TSDB_CODE_SDB_INVALID_DATA_VER;
    goto OFFSET_DECODE_OVER;
  }

  int32_t  size = sizeof(SMqOffsetObj);
  SSdbRow *pRow = sdbAllocRow(size);
  if (pRow == NULL) goto OFFSET_DECODE_OVER;

  SMqOffsetObj *pOffset = sdbGetRowObj(pRow);
  if (pOffset == NULL) goto OFFSET_DECODE_OVER;

  int32_t dataPos = 0;
  int32_t tlen;
  SDB_GET_INT32(pRaw, dataPos, &tlen, OFFSET_DECODE_OVER);
wafwerar's avatar
wafwerar 已提交
118
  buf = taosMemoryMalloc(tlen + 1);
L
Liu Jicong 已提交
119 120 121 122 123 124 125 126 127 128 129
  if (buf == NULL) goto OFFSET_DECODE_OVER;
  SDB_GET_BINARY(pRaw, dataPos, buf, tlen, OFFSET_DECODE_OVER);
  SDB_GET_RESERVE(pRaw, dataPos, MND_OFFSET_RESERVE_SIZE, OFFSET_DECODE_OVER);

  if (tDecodeSMqOffsetObj(buf, pOffset) == NULL) {
    goto OFFSET_DECODE_OVER;
  }

  terrno = TSDB_CODE_SUCCESS;

OFFSET_DECODE_OVER:
wafwerar's avatar
wafwerar 已提交
130
  taosMemoryFreeClear(buf);
L
Liu Jicong 已提交
131 132
  if (terrno != TSDB_CODE_SUCCESS) {
    mError("offset:%s, failed to decode from raw:%p since %s", pOffset->key, pRaw, terrstr());
wafwerar's avatar
wafwerar 已提交
133
    taosMemoryFreeClear(pRow);
L
Liu Jicong 已提交
134 135 136 137 138 139 140
    return NULL;
  }

  mTrace("offset:%s, decode from raw:%p, row:%p", pOffset->key, pRaw, pOffset);
  return pRow;
}

L
Liu Jicong 已提交
141
int32_t mndCreateOffsets(STrans *pTrans, const char *cgroup, const char *topicName, const SArray *vgs) {
L
Liu Jicong 已提交
142 143
  int32_t sz = taosArrayGetSize(vgs);
  for (int32_t i = 0; i < sz; i++) {
L
Liu Jicong 已提交
144
    int32_t      vgId = *(int32_t *)taosArrayGet(vgs, i);
L
Liu Jicong 已提交
145
    SMqOffsetObj offsetObj = {0};
L
Liu Jicong 已提交
146
    if (mndMakePartitionKey(offsetObj.key, cgroup, topicName, vgId) < 0) {
L
Liu Jicong 已提交
147 148
      return -1;
    }
L
Liu Jicong 已提交
149
    // TODO assign db
L
Liu Jicong 已提交
150 151 152 153 154 155 156 157 158 159 160 161 162
    offsetObj.offset = -1;
    SSdbRaw *pOffsetRaw = mndOffsetActionEncode(&offsetObj);
    if (pOffsetRaw == NULL) {
      return -1;
    }
    sdbSetRawStatus(pOffsetRaw, SDB_STATUS_READY);
    if (mndTransAppendRedolog(pTrans, pOffsetRaw) < 0) {
      return -1;
    }
  }
  return 0;
}

S
Shengliang Guan 已提交
163
static int32_t mndProcessCommitOffsetReq(SRpcMsg *pMsg) {
L
Liu Jicong 已提交
164 165
  char key[TSDB_PARTITION_KEY_LEN];

S
Shengliang Guan 已提交
166 167
  SMnode              *pMnode = pMsg->info.node;
  char                *msgStr = pMsg->pCont;
L
Liu Jicong 已提交
168
  SMqCMCommitOffsetReq commitOffsetReq;
H
Hongze Cheng 已提交
169
  SDecoder             decoder;
S
Shengliang Guan 已提交
170
  tDecoderInit(&decoder, msgStr, pMsg->contLen);
L
Liu Jicong 已提交
171 172 173

  tDecodeSMqCMCommitOffsetReq(&decoder, &commitOffsetReq);

S
Shengliang Guan 已提交
174
  STrans *pTrans = mndTransCreate(pMnode, TRN_POLICY_ROLLBACK, TRN_TYPE_COMMIT_OFFSET, pMsg);
L
Liu Jicong 已提交
175 176 177 178 179 180

  for (int32_t i = 0; i < commitOffsetReq.num; i++) {
    SMqOffset *pOffset = &commitOffsetReq.offsets[i];
    if (mndMakePartitionKey(key, pOffset->cgroup, pOffset->topicName, pOffset->vgId) < 0) {
      return -1;
    }
L
Liu Jicong 已提交
181
    bool          create = false;
L
Liu Jicong 已提交
182
    SMqOffsetObj *pOffsetObj = mndAcquireOffset(pMnode, key);
L
Liu Jicong 已提交
183
    if (pOffsetObj == NULL) {
L
Liu Jicong 已提交
184
      pOffsetObj = taosMemoryMalloc(sizeof(SMqOffsetObj));
L
Liu Jicong 已提交
185 186 187
      memcpy(pOffsetObj->key, key, TSDB_PARTITION_KEY_LEN);
      create = true;
    }
L
Liu Jicong 已提交
188 189 190 191
    pOffsetObj->offset = pOffset->offset;
    SSdbRaw *pOffsetRaw = mndOffsetActionEncode(pOffsetObj);
    sdbSetRawStatus(pOffsetRaw, SDB_STATUS_READY);
    mndTransAppendRedolog(pTrans, pOffsetRaw);
L
Liu Jicong 已提交
192 193 194 195 196
    if (create) {
      taosMemoryFree(pOffsetObj);
    } else {
      mndReleaseOffset(pMnode, pOffsetObj);
    }
L
Liu Jicong 已提交
197 198
  }

L
Liu Jicong 已提交
199 200
  tDecoderClear(&decoder);

L
Liu Jicong 已提交
201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222
  if (mndTransPrepare(pMnode, pTrans) != 0) {
    mError("mq-commit-offset-trans:%d, failed to prepare since %s", pTrans->id, terrstr());
    mndTransDrop(pTrans);
    return -1;
  }

  mndTransDrop(pTrans);
  return TSDB_CODE_MND_ACTION_IN_PROGRESS;
}

static int32_t mndOffsetActionInsert(SSdb *pSdb, SMqOffsetObj *pOffset) {
  mTrace("offset:%s, perform insert action", pOffset->key);
  return 0;
}

static int32_t mndOffsetActionDelete(SSdb *pSdb, SMqOffsetObj *pOffset) {
  mTrace("offset:%s, perform delete action", pOffset->key);
  return 0;
}

static int32_t mndOffsetActionUpdate(SSdb *pSdb, SMqOffsetObj *pOldOffset, SMqOffsetObj *pNewOffset) {
  mTrace("offset:%s, perform update action", pOldOffset->key);
L
Liu Jicong 已提交
223
  atomic_store_64(&pOldOffset->offset, pNewOffset->offset);
L
Liu Jicong 已提交
224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244
  return 0;
}

SMqOffsetObj *mndAcquireOffset(SMnode *pMnode, const char *key) {
  SSdb         *pSdb = pMnode->pSdb;
  SMqOffsetObj *pOffset = sdbAcquire(pSdb, SDB_OFFSET, key);
  if (pOffset == NULL && terrno == TSDB_CODE_SDB_OBJ_NOT_THERE) {
    terrno = TSDB_CODE_MND_OFFSET_NOT_EXIST;
  }
  return pOffset;
}

void mndReleaseOffset(SMnode *pMnode, SMqOffsetObj *pOffset) {
  SSdb *pSdb = pMnode->pSdb;
  sdbRelease(pSdb, pOffset);
}

static void mndCancelGetNextOffset(SMnode *pMnode, void *pIter) {
  SSdb *pSdb = pMnode->pSdb;
  sdbCancelFetch(pSdb, pIter);
}
L
Liu Jicong 已提交
245 246 247 248 249 250 251 252 253

static int32_t mndSetDropOffsetCommitLogs(SMnode *pMnode, STrans *pTrans, SMqOffsetObj *pOffset) {
  SSdbRaw *pCommitRaw = mndOffsetActionEncode(pOffset);
  if (pCommitRaw == NULL) return -1;
  if (mndTransAppendCommitlog(pTrans, pCommitRaw) != 0) return -1;
  if (sdbSetRawStatus(pCommitRaw, SDB_STATUS_DROPPED) != 0) return -1;
  return 0;
}

L
Liu Jicong 已提交
254 255 256 257 258 259 260 261
static int32_t mndSetDropOffsetRedoLogs(SMnode *pMnode, STrans *pTrans, SMqOffsetObj *pOffset) {
  SSdbRaw *pRedoRaw = mndOffsetActionEncode(pOffset);
  if (pRedoRaw == NULL) return -1;
  if (mndTransAppendRedolog(pTrans, pRedoRaw) != 0) return -1;
  if (sdbSetRawStatus(pRedoRaw, SDB_STATUS_DROPPED) != 0) return -1;
  return 0;
}

L
Liu Jicong 已提交
262 263 264 265 266 267 268
int32_t mndDropOffsetByDB(SMnode *pMnode, STrans *pTrans, SDbObj *pDb) {
  int32_t code = -1;
  SSdb   *pSdb = pMnode->pSdb;

  void         *pIter = NULL;
  SMqOffsetObj *pOffset = NULL;
  while (1) {
L
Liu Jicong 已提交
269
    pIter = sdbFetch(pSdb, SDB_OFFSET, pIter, (void **)&pOffset);
L
Liu Jicong 已提交
270 271 272 273 274 275 276 277
    if (pIter == NULL) break;

    if (pOffset->dbUid != pDb->uid) {
      sdbRelease(pSdb, pOffset);
      continue;
    }

    if (mndSetDropOffsetCommitLogs(pMnode, pTrans, pOffset) < 0) {
L
Liu Jicong 已提交
278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306
      sdbRelease(pSdb, pOffset);
      goto END;
    }

    sdbRelease(pSdb, pOffset);
  }

  code = 0;
END:
  return code;
}

int32_t mndDropOffsetByTopic(SMnode *pMnode, STrans *pTrans, const char *topic) {
  int32_t code = -1;
  SSdb   *pSdb = pMnode->pSdb;

  void         *pIter = NULL;
  SMqOffsetObj *pOffset = NULL;
  while (1) {
    pIter = sdbFetch(pSdb, SDB_OFFSET, pIter, (void **)&pOffset);
    if (pIter == NULL) break;

    if (!mndOffsetFromTopic(pOffset, topic)) {
      sdbRelease(pSdb, pOffset);
      continue;
    }

    if (mndSetDropOffsetRedoLogs(pMnode, pTrans, pOffset) < 0) {
      sdbRelease(pSdb, pOffset);
L
Liu Jicong 已提交
307 308
      goto END;
    }
L
Liu Jicong 已提交
309 310

    sdbRelease(pSdb, pOffset);
L
Liu Jicong 已提交
311 312 313 314 315 316
  }

  code = 0;
END:
  return code;
}