#4513

Committed 17 Jul 2025 02:02AM UTC coverage: 31.359% (-31.1%) from 62.446%

Build # #4513

Build Type

push

travis-ci

Committed by

web-flow

Commit Message

Merge pull request #31914 from taosdata/fix/3.0/compare-ans-failed

fix:Convert line endings from LF to CRLF for ans file

Run Details

68541 of 301034 branches covered (22.77%)

Branch coverage included in aggregate %.

117356 of 291771 relevant lines covered (40.22%)

602262.98 hits per line

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

1.99

/source/dnode/vnode/src/tsdb/tsdbMigrate.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/>.
 */

#include "tss.h"
#include "tsdb.h"
#include "tsdbFS2.h"
#include "tsdbFSet2.h"
#include "vnd.h"
#include "tsdbInt.h"


extern int32_t tsdbAsyncCompact(STsdb *tsdb, const STimeWindow *tw, bool ssMigrate);


// migrate monitor related functions
typedef struct SSsMigrateMonitor {
  TdThreadCond  stateChanged;
  SVnodeSsMigrateState state;
} SSsMigrateMonitor;


static int32_t getSsMigrateId(STsdb* tsdb) {
  return tsdb->pSsMigrateMonitor->state.vnodeMigrateId;
}


int32_t tsdbOpenSsMigrateMonitor(STsdb *tsdb) {
  SSsMigrateMonitor* pmm = (SSsMigrateMonitor*)taosMemCalloc(1, sizeof(SSsMigrateMonitor));
  if (pmm == NULL) {
    return TSDB_CODE_OUT_OF_MEMORY;
  }

  pmm->state.pFileSetStates = taosArrayInit(16, sizeof(SFileSetSsMigrateState));
  if (pmm->state.pFileSetStates == NULL) {
    taosMemoryFree(pmm);
    return TSDB_CODE_OUT_OF_MEMORY;
  }

  TAOS_UNUSED(taosThreadCondInit(&pmm->stateChanged, NULL));

  pmm->state.dnodeId = vnodeNodeId(tsdb->pVnode);
  pmm->state.vgId = TD_VID(tsdb->pVnode);

  tsdb->pSsMigrateMonitor = pmm;
  return 0;
}


void tsdbCloseSsMigrateMonitor(STsdb *tsdb) {
  SSsMigrateMonitor* pmm = tsdb->pSsMigrateMonitor;
  if (pmm == NULL) {
    return;
  }

  TAOS_UNUSED(taosThreadCondDestroy(&pmm->stateChanged));
  tFreeSVnodeSsMigrateState(&pmm->state);
  taosMemoryFree(tsdb->pSsMigrateMonitor);
  tsdb->pSsMigrateMonitor = NULL;
}

bool tsdbResetSsMigrateMonitor(STsdb *tsdb, int32_t ssMigrateId) {
  SSsMigrateMonitor* pmm = tsdb->pSsMigrateMonitor;
  for (int32_t i = 0; i < taosArrayGetSize(pmm->state.pFileSetStates); i++) {
    SFileSetSsMigrateState *pState = taosArrayGet(pmm->state.pFileSetStates, i);
    if (pState->state == FILE_SET_MIGRATE_STATE_IN_PROGRESS) {
      return false;
    }
  }
  pmm->state.mnodeMigrateId = 0;
  pmm->state.vnodeMigrateId = ssMigrateId;
  pmm->state.startTimeSec = taosGetTimestampSec();
  taosArrayClear(pmm->state.pFileSetStates);
  return true;
}

int32_t tsdbSsMigrateMonitorAddFileSet(STsdb *tsdb, int32_t fid) {
  // no need to lock mutex here, since the caller should have already locked it
  SFileSetSsMigrateState state = { .fid = fid, .state = FILE_SET_MIGRATE_STATE_IN_PROGRESS };
  if (taosArrayPush(tsdb->pSsMigrateMonitor->state.pFileSetStates, &state) == NULL) {
    return terrno;
  }
  return 0;
}

void tsdbSsMigrateMonitorSetFileSetState(STsdb *tsdb, int32_t fid, int32_t state) {
  SSsMigrateMonitor* pmm = tsdb->pSsMigrateMonitor;

  if (taosThreadMutexLock(&tsdb->mutex) != TSDB_CODE_SUCCESS) {
    return;
  }

  for(int32_t i = 0; i < taosArrayGetSize(pmm->state.pFileSetStates); i++) {
    SFileSetSsMigrateState *pState = taosArrayGet(pmm->state.pFileSetStates, i);
    if (pState->fid == fid) {
      pState->state = state;
      break;
    }
  }

  TAOS_UNUSED(taosThreadMutexUnlock(&tsdb->mutex));
}

int32_t tsdbQuerySsMigrateProgress(STsdb *tsdb, int32_t ssMigrateId, int32_t *rspSize, void** ppRsp) {
  int32_t code = 0;

  SSsMigrateMonitor* pmm = tsdb->pSsMigrateMonitor;
  SVnodeSsMigrateState *pState = &pmm->state;

  code = taosThreadMutexLock(&tsdb->mutex);
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }

  pState->mnodeMigrateId = ssMigrateId;
  *rspSize = tSerializeSQuerySsMigrateProgressRsp(NULL, 0, pState);
  if (*rspSize < 0) {
    TAOS_UNUSED(taosThreadMutexUnlock(&tsdb->mutex));
    return TSDB_CODE_INVALID_MSG;
  }

  *ppRsp = rpcMallocCont(*rspSize);
  if (*ppRsp == NULL) {
    TAOS_UNUSED(taosThreadMutexUnlock(&tsdb->mutex));
    tsdbError("rpcMallocCont %d failed", *rspSize);
    return TSDB_CODE_OUT_OF_MEMORY;
  }
  TAOS_UNUSED(tSerializeSQuerySsMigrateProgressRsp(*ppRsp, *rspSize, pState));
  TAOS_UNUSED(taosThreadMutexUnlock(&tsdb->mutex));

  return 0;
}

int32_t tsdbUpdateSsMigrateState(STsdb* tsdb, SVnodeSsMigrateState* pState) {
  int32_t vid = TD_VID(tsdb->pVnode), code = 0;

  // the state was generated by this vnode, so no need to process it
  if (pState->dnodeId == vnodeNodeId(tsdb->pVnode)) {
      tsdbDebug("vgId:%d, skip migration state update since it was generated by this vnode", vid);
      return 0;
  }

  SSsMigrateMonitor* pmm = tsdb->pSsMigrateMonitor;
  SVnodeSsMigrateState *pLocalState = &pmm->state;
  if (pLocalState == NULL) {
    tsdbDebug("vgId:%d, skip migration state update since local state not found", vid);
    return 0;
  }

  code = taosThreadMutexLock(&tsdb->mutex);
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }

  for( int32_t i = 0; i < taosArrayGetSize(pLocalState->pFileSetStates); i++) {
    SFileSetSsMigrateState *pLocalFileSet = taosArrayGet(pLocalState->pFileSetStates, i);
    if (pLocalFileSet->state != FILE_SET_MIGRATE_STATE_IN_PROGRESS) {
      continue; // only update the in-progress file sets
    }

    // a wrong case
    if (pState->mnodeMigrateId != pLocalState->vnodeMigrateId) {
      pLocalFileSet->state = FILE_SET_MIGRATE_STATE_FAILED;
      tsdbError("vgId:%d, fid:%d, set migration state to failed since mnode migrate id mismatch", vid, pLocalFileSet->fid);
      continue;
    }

    // another wrong case
    if (pState->vnodeMigrateId != pLocalState->vnodeMigrateId) {
      pLocalFileSet->state = FILE_SET_MIGRATE_STATE_FAILED;
      tsdbError("vgId:%d, fid:%d, set migration state to failed since vnode migrate id mismatch", vid, pLocalFileSet->fid);
      continue;
    }

    bool found = false;
    for( int32_t j = 0; j < taosArrayGetSize(pState->pFileSetStates); j++) {
      SFileSetSsMigrateState *pRemoteFileSet = taosArrayGet(pState->pFileSetStates, j);
      if (pLocalFileSet->fid == pRemoteFileSet->fid) {
        found = true;
        if (pRemoteFileSet->state != FILE_SET_MIGRATE_STATE_IN_PROGRESS) {
          pLocalFileSet->state = pRemoteFileSet->state;
          tsdbDebug("vgId:%d, fid:%d, migration state was updated to %d", vid, pLocalFileSet->fid, pLocalFileSet->state);
        }
        break;
      }
    }

    // the leader vnode has not this file set, so it will never be migrated, mark it as failed
    // to avoid waiting for forever.
    if (!found) {
      pLocalFileSet->state = FILE_SET_MIGRATE_STATE_FAILED;
      tsdbDebug("vgId:%d, fid:%d, set migration state to failed since remote state not found", vid, pLocalFileSet->fid);
    }
  }

  // always boradcast the state change, so that the waiting threads won't timeout
  TAOS_UNUSED(taosThreadCondBroadcast(&pmm->stateChanged));
  TAOS_UNUSED(taosThreadMutexUnlock(&tsdb->mutex));
  return 0;
}

// migrate file related functions
int32_t tsdbSsFidLevel(int32_t fid, STsdbKeepCfg *pKeepCfg, int32_t ssKeepLocal, int64_t nowSec) {
  int32_t localFid;
  TSKEY   key;

  if (pKeepCfg->precision == TSDB_TIME_PRECISION_MILLI) {
    nowSec = nowSec * 1000;
  } else if (pKeepCfg->precision == TSDB_TIME_PRECISION_MICRO) {
    nowSec = nowSec * 1000000l;
  } else if (pKeepCfg->precision == TSDB_TIME_PRECISION_NANO) {
    nowSec = nowSec * 1000000000l;
  }

  nowSec = nowSec - pKeepCfg->keepTimeOffset * tsTickPerHour[pKeepCfg->precision];

  key = nowSec - ssKeepLocal * tsTickPerMin[pKeepCfg->precision];
  localFid = tsdbKeyFid(key, pKeepCfg->days, pKeepCfg->precision);

  return fid >= localFid ? 0 : 1;
}


static int32_t downloadManifest(SVnode* pVnode, int32_t fid, STFileSet** ppFileSet) {
  int32_t code = 0, vid = TD_VID(pVnode);

  char path[64];
  snprintf(path, sizeof(path), "vnode%d/f%d/manifests.json", vid, fid);
  int64_t size = 0;
  code = tssGetFileSizeOfDefault(path, &size);
  if (code != TSDB_CODE_SUCCESS) {
    tsdbError("vgId:%d, fid:%d, failed to get manifests size since %s", vid, fid, tstrerror(code));
    return code;
  }

  char* buf = taosMemoryMalloc(size + 1);
  code = tssReadFileFromDefault(path, 0, buf, &size);
  if (code != TSDB_CODE_SUCCESS) {
    tsdbError("vgId:%d, fid:%d, failed to read manifest from shared storage since %s", vid, fid, tstrerror(code));
    taosMemoryFree(buf);
    return code;
  }
  buf[size] = 0;

  cJSON* json = cJSON_Parse(buf);
  taosMemoryFree(buf);
  if (json == NULL) {
    tsdbError("vgId:%d, fid:%d, failed to parse manifest json since %s", vid, fid, tstrerror(code));
    return TSDB_CODE_FILE_CORRUPTED;
  }
  
  code = tsdbJsonToTFileSet(pVnode->pTsdb, json, ppFileSet);
  cJSON_Delete(json);
  if (code != TSDB_CODE_SUCCESS) {
    tsdbError("vgId:%d, fid:%d, failed to parse manifest since %s", vid, fid, tstrerror(code));
    return code;
  }

  STFileSet* fset = *ppFileSet;
  if (fset->fid != fid) {
    tsdbError("vgId:%d, fid:%d, mismatch fid, manifest fid is %d", vid, fid, fset->fid);
    tsdbTFileSetClear(ppFileSet);
    return TSDB_CODE_FILE_CORRUPTED;
  }
  if (fset->farr[TSDB_FTYPE_DATA] == NULL) {
    tsdbError("vgId:%d, fid:%d, data file not found in manifest", vid, fid);
    tsdbTFileSetClear(ppFileSet);
    return TSDB_CODE_FILE_CORRUPTED;
  }

  return code;
}


static int32_t uploadManifest(int32_t dnode, int32_t vnode, STFileSet* fset, int32_t mid) {
  int32_t code = 0;

  cJSON* json = cJSON_CreateObject();
  if (json == NULL) {
    return TSDB_CODE_OUT_OF_MEMORY;
  }

  // update migration id for all files in the file set
  STFileObj* fobj = fset->farr[TSDB_FTYPE_HEAD];
  fobj->f->mid = mid;
  fobj = fset->farr[TSDB_FTYPE_SMA];
  fobj->f->mid = mid;
  fobj = fset->farr[TSDB_FTYPE_DATA];
  fobj->f->mid = mid;
  fobj = fset->farr[TSDB_FTYPE_TOMB];
  if (fobj != NULL) {
    fobj->f->mid = mid;
  }

  SSttLvl* lvl;
  TARRAY2_FOREACH(fset->lvlArr, lvl) {
    TARRAY2_FOREACH(lvl->fobjArr, fobj) {
      fobj->f->mid = mid;
    }
  }
  
  if (cJSON_AddNumberToObject(json, "fmtv", 1) == NULL) {
    code = terrno;
    cJSON_Delete(json);
    return code;
  }
  if (cJSON_AddNumberToObject(json, "dnode", dnode) == NULL) {
    code = terrno;
    cJSON_Delete(json);
    return code;
  }
  if (cJSON_AddNumberToObject(json, "vnode", vnode) == NULL) {
    code = terrno;
    cJSON_Delete(json);
    return code;
  }

  code = tsdbTFileSetToJson(fset, json);
  if (code != TSDB_CODE_SUCCESS) {
    cJSON_Delete(json);
    return code;
  }
  char* buf = cJSON_PrintUnformatted(json);
  cJSON_Delete(json);

  char path[64];
  snprintf(path, sizeof(path), "vnode%d/f%d/manifests.json", vnode, fset->fid);
  code = tssUploadToDefault(path, buf, strlen(buf));
  taosMemoryFree(buf);
  if (code != TSDB_CODE_SUCCESS) {
    tsdbError("vgId:%d, fid:%d, failed to upload manifest since %s", vnode, fset->fid, tstrerror(code));
    return code;
  }

  return code;
}


// upload local files to shared storage
//
// local file name is like:
//     [base]/vnode2/f1736/v2f1736ver16.head
// or
//     [base]/vnode2/f1736/v2f1736ver16.m334233.head
//
// remote file name is like:
//     vnode2/f1736/v2f1736ver16.m13552343.head
//
// NOTE: the migration id is always included in the remote file name, because
// the commit id may be different between the vnodes of the same vgroup,
// that's an interrupted migration may overwrite some of the remote files
// while leaving the others intact if we don't include the migration id in
// the remote file name.
static int32_t uploadFile(SRTNer* rtner, STFileObj* fobj) {
  if (fobj == NULL) {
    return TSDB_CODE_SUCCESS;
  }

  const char* ext = strrchr(fobj->fname, '.');
  int32_t vid = TD_VID(rtner->tsdb->pVnode), mid = getSsMigrateId(rtner->tsdb);
  STFile* f = fobj->f;
  
  char path[TSDB_FILENAME_LEN];
  snprintf(path, sizeof(path), "vnode%d/f%d/v%df%dver%" PRId64 ".m%d%s", vid, f->fid, vid, f->fid, f->cid, mid, ext);

  int code = tssUploadFileToDefault(path, fobj->fname, 0, -1);
  if (code != TSDB_CODE_SUCCESS) {
    tsdbError("vgId:%d, fid:%d, failed to upload file %s since %s", vid, f->fid, fobj->fname, tstrerror(code));
    return code;
  }

  return 0;
}



static int32_t downloadFile(SRTNer* rtner, STFileObj* fobj) {
  if (fobj == NULL) {
    return TSDB_CODE_SUCCESS;
  }

  const char* fname = strrchr(fobj->fname, TD_DIRSEP_CHAR) + 1;
  int32_t vid = TD_VID(rtner->tsdb->pVnode);
  STFile* f = fobj->f;
  
  char path[TSDB_FILENAME_LEN];
  snprintf(path, sizeof(path), "vnode%d/f%d/%s", vid, f->fid, fname);
  int code = tssDownloadFileFromDefault(path, fobj->fname, 0, -1);
  if (code != TSDB_CODE_SUCCESS) {
    tsdbError("vgId:%d, fid:%d, failed to download file %s since %s", vid, f->fid, path, tstrerror(code));
    return code;
  }

  return 0;
}


static void tsdbRemoveSsGarbageFiles(int32_t vid, STFileSet* fset) {
  char prefix[TSDB_FILENAME_LEN];
  snprintf(prefix, sizeof(prefix), "vnode%d/f%d/", vid, fset->fid);

  SArray* paths = taosArrayInit(10, sizeof(char*));
  int32_t code = tssListFileOfDefault(prefix, paths);
  if (code != TSDB_CODE_SUCCESS) {
    tsdbError("vgId:%d, fid:%d, failed to list files in shared storage since %s", vid, fset->fid, tstrerror(code));
    taosArrayDestroy(paths);
    return;
  }

  for(int i = 0; i < taosArrayGetSize(paths); i++) {
      char* p = *(char**)taosArrayGet(paths, i);
      const char* ext = strrchr(p, '.');
      const char* rname = strrchr(p, '/') + 1;
      bool remove = true;
      int32_t vgId = 0, fid = 0, mid = 0, cn = 0;
      int64_t cid = 0;
  
      // NOTE: when compare file name, don't use strcmp(fobj->fname, rname) because 'fobj->fname' may not
      // contain the migration id. that's why we use sscanf to parse the file name.

      if (ext == NULL) {
        // no extension, remove
      } else if (taosStrcasecmp(ext, ".head") == 0) {
        STFileObj* fobj = fset->farr[TSDB_FTYPE_HEAD];
        int n = sscanf(rname, "v%df%dver%" PRId64 ".m%d.head", &vgId, &fid, &cid, &mid);
        remove = (n != 4 || vgId != vid || fid != fset->fid || cid != fobj->f->cid || mid != fobj->f->mid);
      } else if (taosStrcasecmp(ext, ".sma") == 0) {
        STFileObj* fobj = fset->farr[TSDB_FTYPE_SMA];
        int n = sscanf(rname, "v%df%dver%" PRId64 ".m%d.sma", &vgId, &fid, &cid, &mid);
        remove = (n != 4 || vgId != vid || fid != fset->fid || cid != fobj->f->cid || mid != fobj->f->mid);
      } else if (taosStrcasecmp(ext, ".tomb") == 0) {
        STFileObj* fobj = fset->farr[TSDB_FTYPE_TOMB];
        if (fobj) {
          int n = sscanf(rname, "v%df%dver%" PRId64 ".m%d.tomb", &vgId, &fid, &cid, &mid);
          remove = (n != 4 || vgId != vid || fid != fset->fid || cid != fobj->f->cid || mid != fobj->f->mid);
        }
      } else if (taosStrcasecmp(ext, ".stt") == 0) {
        SSttLvl* lvl = NULL;
        TARRAY2_FOREACH(fset->lvlArr, lvl) {
          STFileObj* fobj;
          TARRAY2_FOREACH(lvl->fobjArr, fobj) {
            int n = sscanf(rname, "v%df%dver%" PRId64 ".m%d.stt", &vgId, &fid, &cid, &mid);
            if (n == 4 && vgId == vid && fid == fset->fid && cid == fobj->f->cid && mid == fobj->f->mid) {
              remove = false;
              break;
            }
          }
          if (!remove) {
            break;
          }
        }
      } else if (taosStrcasecmp(ext, ".data") == 0) {
        STFileObj* fobj = fset->farr[TSDB_FTYPE_DATA];
        int n = sscanf(rname, "v%df%dver%" PRId64 ".%d.data", &vgId, &fid, &cid, &cn);
        if (n == 4) {
          if (vgId == vid && fid == fset->fid && cid == fobj->f->cid && cn >= 1 && cn < fobj->f->lcn) {
            remove = false; // not the last chunk, keep it
          }
        } else {
          n = sscanf(rname, "v%df%dver%" PRId64 ".m%d.%d.data", &vgId, &fid, &cid, &mid, &cn);
          remove = (n != 5 || vgId != vid || fid != fset->fid || cid != fobj->f->cid || mid != fobj->f->mid || cn != fobj->f->lcn);
        }
      } else {
        remove = (taosStrcasecmp(rname, "manifests.json") != 0); // keep manifest, remove all other files
      }

      if (remove) {
        int32_t code = tssDeleteFileFromDefault(p);
        if (code != TSDB_CODE_SUCCESS) {
          tsdbError("vgId:%d, fid:%d, failed to remove garbage file %s from shared storage since %s", vid, fset->fid, p, tstrerror(code));
        } else {
          tsdbInfo("vgId:%d, fid:%d, garbage file %s is removed from shared storage", vid, fset->fid, p);
        }
      }

      taosMemoryFree(p);
  }

  taosArrayDestroy(paths);
}


// download the last chunk of a data file
// remote file name is like:
//      vnode2/f1736/v2f1736ver16.m13552343.4.data
static int32_t downloadDataFileLastChunk(SRTNer* rtner, STFileObj* fobj) {
  int32_t code = 0;
  int32_t vid = TD_VID(rtner->tsdb->pVnode);
  SVnodeCfg *pCfg = &rtner->tsdb->pVnode->config;
  STFile *f = fobj->f;

  char lpath[TSDB_FILENAME_LEN], rpath[TSDB_FILENAME_LEN];
  tsdbTFileLastChunkName(rtner->tsdb, f, lpath);
  char* fname = strrchr(lpath, TD_DIRSEP_CHAR) + 1;

  sprintf(rpath, "vnode%d/f%d/%s", vid, f->fid, fname);

  code = tssDownloadFileFromDefault(rpath, lpath, 0, -1);
  if (code != TSDB_CODE_SUCCESS) {
    tsdbError("vgId:%d, fid:%d, failed to download data file %s since %s", vid, f->fid, rpath, tstrerror(code));
    return code;
  }

  return code;
}


// while other files all include the migration id in the remote file name, only the last
// chunk of a data file does the same. this is ok because:
// 1. without a compaction, data file is always uploaded chunk by chunk, only the last
//    chunk may be modified.
// 2. after a compaction, all of the data is downloaded to local, so overwriting remote
//    data chunks won't cause any problem. (this is not likely to happen because we will
//    cancel the migration in this case, refer comment in function shouldMigrate).
static int32_t uploadDataFile(SRTNer* rtner, STFileObj* fobj) {
  int32_t code = 0;
  int32_t vid = TD_VID(rtner->tsdb->pVnode), mid = getSsMigrateId(rtner->tsdb);
  SVnodeCfg *pCfg = &rtner->tsdb->pVnode->config;
  int64_t szFile = 0, szChunk = (int64_t)pCfg->tsdbPageSize * pCfg->ssChunkSize;
  STFile *f = fobj->f;
  STFileOp op = {.optype = TSDB_FOP_MODIFY, .fid = f->fid, .of = *f};

  char path[TSDB_FILENAME_LEN];
  if (f->lcn <= 1) {
    strcpy(path, fobj->fname);
  } else {
    tsdbTFileLastChunkName(rtner->tsdb, f, path);
  }

  code = taosStatFile(path, &szFile, NULL, NULL);
  if (code != TSDB_CODE_SUCCESS) {
    tsdbError("vgId:%d, fid:%d failed to stat file %s since %s", vid, f->fid, path, tstrerror(code));
    return false;
  }

  if (f->lcn > 1) {
    szFile += szChunk * (f->lcn - 1); // add the size of migrated chunks
  }

  int totalChunks = szFile / szChunk;
  if (szFile % szChunk) {
    totalChunks++;
  }

  int lcn = f->lcn < 1 ? 1 : f->lcn;

  // upload chunks one by one, the first chunk may already been uploaded, but may be
  // modified thereafter, so we need to upload it again
  for (int i = lcn; i <= totalChunks; ++i) {
    int64_t offset = (int64_t)(i - lcn) * szChunk;
    int64_t size = szChunk;
    if (i == totalChunks && szFile % szChunk) {
        size = szFile % szChunk;
    }

    // only include the migration id in the last chunk filename
    char rpath[TSDB_FILENAME_LEN];
    if (i == totalChunks) {
      sprintf(rpath, "vnode%d/f%d/v%df%dver%" PRId64 ".m%d.%d.data", vid, f->fid, vid, f->fid, f->cid, mid, i);
    } else {
      sprintf(rpath, "vnode%d/f%d/v%df%dver%" PRId64 ".%d.data", vid, f->fid, vid, f->fid, f->cid, i);
    }

    code = tssUploadFileToDefault(rpath, path, offset, size);
    if (code != TSDB_CODE_SUCCESS) {
      tsdbError("vgId:%d, fid:%d, failed to migrate data file since %s", vid, f->fid, tstrerror(code));
      return code;
    }
  }

  f->lcn = totalChunks;
  op.nf = *f;
  code = TARRAY2_APPEND(&rtner->fopArr, op);
  if (code != TSDB_CODE_SUCCESS) {
    tsdbError("vgId:%d, fid:%d, failed to append file operation since %s", vid, f->fid, tstrerror(code));
    return code;
  }

  // manifest must be uploaded before copy last chunk, otherwise, failed to upload manifest
  // will result in a broken migration
  tsdbInfo("vgId:%d, fid:%d, data file migrated, begin generate & upload manifest", vid, f->fid);

  // manifest, this also commit the migration
  code = uploadManifest(vnodeNodeId(rtner->tsdb->pVnode), vid, rtner->fset, getSsMigrateId(rtner->tsdb));
  if (code != TSDB_CODE_SUCCESS) {
    tsdbSsMigrateMonitorSetFileSetState(rtner->tsdb, f->fid, FILE_SET_MIGRATE_STATE_FAILED);
    return code;
  }

  tsdbInfo("vgId:%d, fid:%d, manifest uploaded, begin remove garbage files", vid, f->fid);
  tsdbRemoveSsGarbageFiles(vid, rtner->fset);

  tsdbSsMigrateMonitorSetFileSetState(rtner->tsdb, f->fid, FILE_SET_MIGRATE_STATE_SUCCEEDED);
  tsdbInfo("vgId:%d, fid:%d, leader migration succeeded", vid, f->fid);

  // no new chunks generated, no need to copy the last chunk
  if (totalChunks == lcn) {
    return 0;
  }

  // copy the last chunk to the new file
  char newPath[TSDB_FILENAME_LEN];
  tsdbTFileLastChunkName(rtner->tsdb, &op.nf, newPath);

  int64_t offset = (int64_t)(totalChunks - lcn) * szChunk;
  int64_t size = szChunk;
  if (szFile % szChunk) {
    size = szFile % szChunk;
  }

  TdFilePtr fdFrom = taosOpenFile(path, TD_FILE_READ);
  if (fdFrom == NULL) {
    code = terrno;
    tsdbError("vgId:%d, fid:%d, failed to open source file %s since %s", vid, f->fid, path, tstrerror(code));
    return code;
  }

  TdFilePtr fdTo = taosOpenFile(newPath, TD_FILE_WRITE | TD_FILE_CREATE | TD_FILE_TRUNC);
  if (fdTo == NULL) {
    code = terrno;
    tsdbError("vgId:%d, fid:%d, failed to open target file %s since %s", vid, f->fid, newPath, tstrerror(code));
    TAOS_UNUSED(taosCloseFile(&fdFrom));
    return code;
  }

  int64_t n = taosFSendFile(fdTo, fdFrom, &offset, size);
  if (n < 0) {
    code = terrno;
    tsdbError("vgId:%d, fid:%d, failed to copy file %s to %s since %s", vid, f->fid, path, newPath, tstrerror(code));
  }
  TAOS_UNUSED(taosCloseFile(&fdFrom));
  TAOS_UNUSED(taosCloseFile(&fdTo));

  return code;
}


static bool shouldMigrate(SRTNer *rtner, int32_t *pCode) {
  int32_t vid = TD_VID(rtner->tsdb->pVnode);
  SVnodeCfg *pCfg = &rtner->tsdb->pVnode->config;
  STFileSet *pLocalFset = rtner->fset;
  STFileObj *flocal = pLocalFset->farr[TSDB_FTYPE_DATA];

  *pCode = 0;
  if (!flocal) {
    tsdbInfo("vgId:%d, fid:%d, migration cancelled, local data file not exist", vid, pLocalFset->fid);
    return false;
  }

  if (rtner->lastCommit != pLocalFset->lastCommit) {
    tsdbInfo("vgId:%d, fid:%d, migration cancelled, there are new commits after migration task is scheduled", vid, pLocalFset->fid);
    return false;
  }

  if (pCfg->ssCompact && flocal->f->lcn < 0) {
    int32_t lcn = flocal->f->lcn;
    STimeWindow win = {0};
    tsdbFidKeyRange(pLocalFset->fid, rtner->tsdb->keepCfg.days, rtner->tsdb->keepCfg.precision, &win.skey, &win.ekey);
    *pCode = tsdbAsyncCompact(rtner->tsdb, &win, true);
    tsdbInfo("vgId:%d, fid:%d, migration cancelled, fileset need compact, lcn: %d", vid, pLocalFset->fid, lcn);
    return false; // compact in progress
  }

  char path[TSDB_FILENAME_LEN];
  if (flocal->f->lcn <= 1) {
    strcpy(path, flocal->fname);
  } else {
    tsdbTFileLastChunkName(rtner->tsdb, flocal->f, path);
  }

  int64_t mtime = 0, size = 0;
  *pCode = taosStatFile(path, &size, &mtime, NULL);
  if (*pCode != TSDB_CODE_SUCCESS) {
    tsdbError("vgId:%d, fid:%d, migration cancelled, failed to stat file %s since %s", vid, pLocalFset->fid, path, tstrerror(*pCode));
    return false;
  }

  if (size <= (int64_t)pCfg->tsdbPageSize * pCfg->ssChunkSize) {
    tsdbInfo("vgId:%d, fid:%d, migration skipped, data file is too small, size: %" PRId64 " bytes", vid, pLocalFset->fid, size);
    return false; // file too small, no need to migrate
  }

  if (mtime >= rtner->now - tsSsUploadDelaySec) {
    tsdbInfo("vgId:%d, fid:%d, migration skipped, data file is active writting, modified at %" PRId64, vid, pLocalFset->fid, mtime);
    return false; // still active writing, postpone migration
  }

  if (tsdbFidLevel(pLocalFset->fid, &rtner->tsdb->keepCfg, rtner->now) < 0) {
    tsdbInfo("vgId:%d, fid:%d, migration skipped, file set is expired", vid, pLocalFset->fid);
    return false; // file set expired
  }

  if (tsdbSsFidLevel(pLocalFset->fid, &rtner->tsdb->keepCfg, pCfg->ssKeepLocal, rtner->now) < 1) {
    tsdbInfo("vgId:%d, fid:%d, migration skipped, keep local file set", vid, pLocalFset->fid);
    return false; // keep on local storage
  }

  // download manifest from shared storage
  STFileSet *pRemoteFset = NULL;
  *pCode = downloadManifest(rtner->tsdb->pVnode, pLocalFset->fid, &pRemoteFset);
  if (*pCode == TSDB_CODE_SUCCESS) {
    // remote file exists but local file has not been migrated, there are two possibilities:
    // 1. there's a compact after the last migration, this is a normal case, we can discard
    //    the remote files and continue the migration;
    // 2. in the last migration, this node was a follower, the leader did its job successfully,
    //    but this node did not, continue the migration may overwrite remote file and result in
    //    data corruption on other nodes.
    // however, it is hard to distinguish them, so just treat both as a migration error. hope
    // the user could do something to recover, such as remove remote files.
    if (flocal->f->lcn < 1) {
      tsdbTFileSetClear(&pRemoteFset);
      tsdbError("vgId:%d, fid:%d, migration cancelled, remote manifest found but local lcn < 1", vid, pLocalFset->fid);
      *pCode = TSDB_CODE_FILE_CORRUPTED;
      return false;
    }

  } else if (*pCode == TSDB_CODE_NOT_FOUND) {
    if (flocal->f->lcn >= 1) {
      tsdbError("vgId:%d, fid:%d, migration cancelled, remote manifest not found but local lcn >= 1", vid, pLocalFset->fid);
      *pCode = TSDB_CODE_FILE_CORRUPTED;
      return false;
    }

    // this is the first migration, we should do it.
    *pCode = TSDB_CODE_SUCCESS;
    return true;

  } else {
    tsdbError("vgId:%d, fid:%d, migration cancelled, failed to download manifest, code: %d", vid, pLocalFset->fid, *pCode);
    return false;
  }

  STFileObj *fremote = pRemoteFset->farr[TSDB_FTYPE_DATA];
  if (fremote == NULL) {
    tsdbError("vgId:%d, fid:%d, migration cancelled, cannot find data file information from remote manifest", vid, pLocalFset->fid);
    tsdbTFileSetClear(&pRemoteFset);
    *pCode = TSDB_CODE_FILE_CORRUPTED;
    return false;
  }

  if (fremote->f->lcn != flocal->f->lcn) {
    tsdbError("vgId:%d, fid:%d, migration cancelled, remote and local data file information mismatch", vid, pLocalFset->fid);
    tsdbTFileSetClear(&pRemoteFset);
    *pCode = TSDB_CODE_FILE_CORRUPTED;
    return false;
  }
  
  if (fremote->f->maxVer == flocal->f->maxVer) {
    tsdbTFileSetClear(&pRemoteFset);
    tsdbError("vgId:%d, fid:%d, migration skipped, no new data", vid, pLocalFset->fid);
    return false; // no new data
  }

  tsdbTFileSetClear(&pRemoteFset); // we use the local file set information for migration
  tsdbInfo("vgId:%d, fid:%d, file set will be migrated", vid, pLocalFset->fid);
  return true;
}


static int32_t tsdbFollowerDoSsMigrate(SRTNer *rtner) {
  int32_t code = 0, vid = TD_VID(rtner->tsdb->pVnode);
  STFileSet *fset = rtner->fset;
  SSsMigrateMonitor* pmm = rtner->tsdb->pSsMigrateMonitor;
  SFileSetSsMigrateState *pState = NULL;
  int32_t fsIdx = 0;

  // though we make this check in the leader node, we should do this in the follower nodes too.
  // because there may be a leader change and the execution order of async tasks may result in
  // different commit time. if we don't do this, we may corrupt the follower data.
  if (rtner->lastCommit != fset->lastCommit) {
    tsdbInfo("vgId:%d, fid:%d, follower migration cancelled, there are new commits after migration is scheduled", vid, fset->fid);
    return 0;
  }

  tsdbInfo("vgId:%d, fid:%d, vnode is follower, waiting leader on node %d to upload.", vid, fset->fid, rtner->nodeId);

  code = taosThreadMutexLock(&rtner->tsdb->mutex);
  if (code != TSDB_CODE_SUCCESS) {
    return code;
  }

  for (; fsIdx < taosArrayGetSize(pmm->state.pFileSetStates); fsIdx++) {
    pState = taosArrayGet(pmm->state.pFileSetStates, fsIdx);
    if (pState->fid == fset->fid) {
      break;
    }
  }

  while(pState->state == FILE_SET_MIGRATE_STATE_IN_PROGRESS) {
    struct timespec ts;
    if ((code = taosClockGetTime(CLOCK_REALTIME, &ts)) != TSDB_CODE_SUCCESS) {
      tsdbError("vgId:%d, fid:%d, failed to get current time since %s", vid, fset->fid, tstrerror(code));
      TAOS_UNUSED(taosThreadMutexUnlock(&rtner->tsdb->mutex));
      return code;
    }
    ts.tv_sec += 30; // TODO: make it configurable
    code = taosThreadCondTimedWait(&pmm->stateChanged, &rtner->tsdb->mutex, &ts);
    pState = taosArrayGet(pmm->state.pFileSetStates, fsIdx);
    if (code == TSDB_CODE_TIMEOUT_ERROR) {
      tsdbError("vgId:%d, fid:%d, waiting leader migration timed out", vid, fset->fid);
      pState->state = FILE_SET_MIGRATE_STATE_FAILED;
    }
  }

  TAOS_UNUSED(taosThreadMutexUnlock(&rtner->tsdb->mutex));

  if (pState->state != FILE_SET_MIGRATE_STATE_SUCCEEDED) {
    tsdbInfo("vgId:%d, fid:%d, follower migration skipped because leader migration skipped or failed", vid, fset->fid);
    return 0;
  }

  tsdbInfo("vgId:%d, fid:%d, follower migration started, begin downloading manifest...", vid, fset->fid);
  STFileSet *pRemoteFset = NULL;
  code = downloadManifest(rtner->tsdb->pVnode, fset->fid, &pRemoteFset);
  if (code == TSDB_CODE_NOT_FOUND) {
    tsdbTFileSetClear(&pRemoteFset);
    return TSDB_CODE_FILE_CORRUPTED;
  }

  // this often happens in the catch up process of a new node, it is ok to continue, but will
  // result in download same files more than once, which is a waste of time and bandwidth.
  if (pRemoteFset->farr[TSDB_FTYPE_HEAD]->f->mid != getSsMigrateId(rtner->tsdb)) {
    tsdbTFileSetClear(&pRemoteFset);
    tsdbWarn("vgId:%d, fid:%d, follower migration cancelled, migration id mismatch", vid, fset->fid);
    return 0;
  }

  tsdbInfo("vgId:%d, fid:%d, manifest downloaded, begin downloading head file", vid, fset->fid);
  code = downloadFile(rtner, pRemoteFset->farr[TSDB_FTYPE_HEAD]);
  if (code != TSDB_CODE_SUCCESS) {
    tsdbTFileSetClear(&pRemoteFset);
    return code;
  }
  STFileOp op = {.optype = TSDB_FOP_MODIFY, .fid = fset->fid, .of = *fset->farr[TSDB_FTYPE_HEAD]->f, .nf = *pRemoteFset->farr[TSDB_FTYPE_HEAD]->f};
  code = TARRAY2_APPEND(&rtner->fopArr, op);
  if (code != TSDB_CODE_SUCCESS) {
    tsdbTFileSetClear(&pRemoteFset);
    tsdbError("vgId:%d, fid:%d, failed to append head file operation since %s", vid, fset->fid, tstrerror(code));
    return code;
  }

  tsdbInfo("vgId:%d, fid:%d, head file downloaded, begin downloading sma file", vid, fset->fid);
  code = downloadFile(rtner, pRemoteFset->farr[TSDB_FTYPE_SMA]);
  if (code != TSDB_CODE_SUCCESS) {
    tsdbTFileSetClear(&pRemoteFset);
    return code;
  }
  op = (STFileOp) {.optype = TSDB_FOP_MODIFY, .fid = fset->fid, .of = *fset->farr[TSDB_FTYPE_SMA]->f, .nf = *pRemoteFset->farr[TSDB_FTYPE_SMA]->f};
  code = TARRAY2_APPEND(&rtner->fopArr, op);
  if (code != TSDB_CODE_SUCCESS) {
    tsdbTFileSetClear(&pRemoteFset);
    tsdbError("vgId:%d, fid:%d, failed to append sma file operation since %s", vid, fset->fid, tstrerror(code));
    return code;
  }

  tsdbInfo("vgId:%d, fid:%d, sma file downloaded, begin downloading tomb file", vid, fset->fid);
  code = downloadFile(rtner, pRemoteFset->farr[TSDB_FTYPE_TOMB]);
  if (code != TSDB_CODE_SUCCESS) {
    tsdbTFileSetClear(&pRemoteFset);
    return code;
  }
  if (fset->farr[TSDB_FTYPE_TOMB] != NULL && pRemoteFset->farr[TSDB_FTYPE_TOMB] != NULL) {
    op = (STFileOp) {.optype = TSDB_FOP_MODIFY, .fid = fset->fid, .of = *fset->farr[TSDB_FTYPE_TOMB]->f, .nf = *pRemoteFset->farr[TSDB_FTYPE_TOMB]->f};
    code = TARRAY2_APPEND(&rtner->fopArr, op);
  } else if (fset->farr[TSDB_FTYPE_TOMB] != NULL) {
    // the remote tomb file is not found, but local tomb file exists, we should remove it
    op = (STFileOp) {.optype = TSDB_FOP_REMOVE, .fid = fset->fid, .of = *fset->farr[TSDB_FTYPE_TOMB]->f};
    code = TARRAY2_APPEND(&rtner->fopArr, op);
  } else if (pRemoteFset->farr[TSDB_FTYPE_TOMB] != NULL) {
    op = (STFileOp) {.optype = TSDB_FOP_CREATE, .fid = fset->fid, .nf = *pRemoteFset->farr[TSDB_FTYPE_TOMB]->f};
    code = TARRAY2_APPEND(&rtner->fopArr, op);
  }
  if (code != TSDB_CODE_SUCCESS) {
    tsdbTFileSetClear(&pRemoteFset);
    tsdbError("vgId:%d, fid:%d, failed to append tomb file operation since %s", vid, fset->fid, tstrerror(code));
    return code;
  }

  tsdbInfo("vgId:%d, fid:%d, tomb file downloaded, begin downloading stt files", vid, fset->fid);
  SSttLvl* lvl;
  TARRAY2_FOREACH(fset->lvlArr, lvl) {
    STFileObj* fobj;
    TARRAY2_FOREACH(lvl->fobjArr, fobj) {
      op = (STFileOp) {.optype = TSDB_FOP_REMOVE, .fid = fset->fid, .of = *fobj->f};
      code = TARRAY2_APPEND(&rtner->fopArr, op);
      if (code != TSDB_CODE_SUCCESS) {
        tsdbTFileSetClear(&pRemoteFset);
        tsdbError("vgId:%d, fid:%d, failed to append stt file remove operation since %s", vid, fset->fid, tstrerror(code));
        return code;
      }
    }
  }
  TARRAY2_FOREACH(pRemoteFset->lvlArr, lvl) {
    STFileObj* fobj;
    TARRAY2_FOREACH(lvl->fobjArr, fobj) {
      code = downloadFile(rtner, fobj);
      if (code != TSDB_CODE_SUCCESS) {
        tsdbTFileSetClear(&pRemoteFset);
        return code;
      }
      op = (STFileOp) {.optype = TSDB_FOP_CREATE, .fid = fset->fid, .nf = *fobj->f};
      code = TARRAY2_APPEND(&rtner->fopArr, op);
      if (code != TSDB_CODE_SUCCESS) {
        tsdbTFileSetClear(&pRemoteFset);
        tsdbError("vgId:%d, fid:%d, failed to append stt file create operation since %s", vid, fset->fid, tstrerror(code));
        return code;
      }
    }
  }

  tsdbInfo("vgId:%d, fid:%d, stt files downloaded, begin downloading data file", vid, fset->fid);
  code = downloadDataFileLastChunk(rtner, pRemoteFset->farr[TSDB_FTYPE_DATA]);
  if (code != TSDB_CODE_SUCCESS) {
    tsdbTFileSetClear(&pRemoteFset);
    return code;
  }
  op = (STFileOp) {.optype = TSDB_FOP_MODIFY, .fid = fset->fid, .of = *fset->farr[TSDB_FTYPE_DATA]->f, .nf = *pRemoteFset->farr[TSDB_FTYPE_DATA]->f};
  code = TARRAY2_APPEND(&rtner->fopArr, op);
  if (code != TSDB_CODE_SUCCESS) {
    tsdbTFileSetClear(&pRemoteFset);
    tsdbError("vgId:%d, fid:%d, failed to append data file operation since %s", vid, fset->fid, tstrerror(code));
    return code;
  }

  tsdbInfo("vgId:%d, fid:%d, data file downloaded", vid, fset->fid);
  tsdbTFileSetClear(&pRemoteFset);
  return 0;
}


static int32_t tsdbLeaderDoSsMigrate(SRTNer *rtner) {
  int32_t code = 0, vid = TD_VID(rtner->tsdb->pVnode);
  SVnodeCfg *pCfg = &rtner->tsdb->pVnode->config;
  STFileSet *fset = rtner->fset;

  tsdbInfo("vgId:%d, fid:%d, vnode is leader, migration started", vid, fset->fid);

  if (!shouldMigrate(rtner, &code)) {
    int32_t state = (code == TSDB_CODE_SUCCESS) ? FILE_SET_MIGRATE_STATE_SKIPPED : FILE_SET_MIGRATE_STATE_FAILED;
    tsdbSsMigrateMonitorSetFileSetState(rtner->tsdb, fset->fid, state);
    return code;
  }

  // head file
  tsdbInfo("vgId:%d, fid:%d, begin migrate head file", vid, fset->fid);
  code = uploadFile(rtner, fset->farr[TSDB_FTYPE_HEAD]);
  if (code != TSDB_CODE_SUCCESS) {
    tsdbSsMigrateMonitorSetFileSetState(rtner->tsdb, fset->fid, FILE_SET_MIGRATE_STATE_FAILED);
    return code;
  }

  tsdbInfo("vgId:%d, fid:%d, head file migrated, begin migrate sma file", vid, fset->fid);

  // sma file
  code = uploadFile(rtner, fset->farr[TSDB_FTYPE_SMA]);
  if (code != TSDB_CODE_SUCCESS) {
    tsdbSsMigrateMonitorSetFileSetState(rtner->tsdb, fset->fid, FILE_SET_MIGRATE_STATE_FAILED);
    return code;
  }

  tsdbInfo("vgId:%d, fid:%d, sma file migrated, begin migrate tomb file", vid, fset->fid);

  // tomb file
  code = uploadFile(rtner, fset->farr[TSDB_FTYPE_TOMB]);
  if (code != TSDB_CODE_SUCCESS) {
    tsdbSsMigrateMonitorSetFileSetState(rtner->tsdb, fset->fid, FILE_SET_MIGRATE_STATE_FAILED);
    return code;
  }

  tsdbInfo("vgId:%d, fid:%d, tomb file migrated, begin migrate stt files", vid, fset->fid);

  // stt files
  SSttLvl* lvl;
  TARRAY2_FOREACH(fset->lvlArr, lvl) {
    STFileObj* fobj;
    TARRAY2_FOREACH(lvl->fobjArr, fobj) {
      code = uploadFile(rtner, fobj);
      if (code != TSDB_CODE_SUCCESS) {
        tsdbSsMigrateMonitorSetFileSetState(rtner->tsdb, fset->fid, FILE_SET_MIGRATE_STATE_FAILED);
        return code;
      }
    }
  }
  
  tsdbInfo("vgId:%d, fid:%d, stt files migrated, begin migrate data file", vid, fset->fid);

  // data file
  code = uploadDataFile(rtner, fset->farr[TSDB_FTYPE_DATA]);
  if (code != TSDB_CODE_SUCCESS) {
    tsdbSsMigrateMonitorSetFileSetState(rtner->tsdb, fset->fid, FILE_SET_MIGRATE_STATE_FAILED);
    return code;
  }

  return TSDB_CODE_SUCCESS;
}


int32_t tsdbDoSsMigrate(SRTNer *rtner) {
  // note: leader is decided when the task is scheduled, the actual leader may change after that,
  // but this is ok.
  if (rtner->nodeId == vnodeNodeId(rtner->tsdb->pVnode)) {
    return tsdbLeaderDoSsMigrate(rtner);
  }
  return tsdbFollowerDoSsMigrate(rtner);
}

taosdata / TDengine / #4513

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

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