lib/cargo.js
// commit local rotated files to remote ClickHouse DB
commit local rotated files to remote ClickHouse DB
commit local rotated files to remote ClickHouse DB
async commitToClickhouseDB() {
      var err, filenamList, rotationPrefix, splitMark;
      try {
        filenamList = (await fsAsync.readdir(this.pathToCargoFolder));
      } catch (error) {
        err = error;
        debuglog("[commitToClickhouseDB > ls] FAILED error:", err);
        return;
      }
      if (!(Array.isArray(filenamList) && (filenamList.length > 0))) {
        return;
      }
      rotationPrefix = FILENAME_PREFIX + this.tableName + '.';
      filenamList = filenamList.filter(function(item) {
        return item.startsWith(rotationPrefix) && item.endsWith(EXTNAME_UNCOMMITTED);
      });
      if (!(filenamList.length > 0)) {
        return;
      }
      debuglog(`[commitToClickhouseDB] filenamList(${filenamList.length})`);
      splitMark = `${FILENAME_PREFIX}${this.tableName}.`;
      filenamList.sort(function(a, b) {
        var tsA, tsB;
        tsA = parseInt((a.split(splitMark) || [])[1] || 0, 36);
        tsB = parseInt((b.split(splitMark) || [])[1] || 0, 36);
        return tsB - tsA;
      });
      filenamList.length = Math.min(filenamList.length, this.maxInsetParts);
      filenamList = filenamList.map((item) => {
        return path.join(this.pathToCargoFolder, item);
      });
      this.uploadLocalUncommitsToClickhouse(filenamList);
    }
async commitToClickhouseDB ( ) { var err , filenamList , rotationPrefix , splitMark ; try { filenamList = ( await fsAsync . readdir ( this . pathToCargoFolder ) ) ; } catch ( error ) { err = error ; debuglog ( "[commitToClickhouseDB > ls] FAILED error:" , err ) ; return ; } if ( ! ( Array . isArray ( filenamList ) && ( filenamList . length > 0 ) ) ) { return ; } rotationPrefix = FILENAME_PREFIX + this . tableName + '.' ; filenamList = filenamList . filter ( function ( item ) { return item . startsWith ( rotationPrefix ) && item . endsWith ( EXTNAME_UNCOMMITTED ) ; } ) ; if ( ! ( filenamList . length > 0 ) ) { return ; } debuglog ( ` ${ filenamList . length } ` ) ; splitMark = ` ${ FILENAME_PREFIX } ${ this . tableName } ` ; filenamList . sort ( function ( a , b ) { var tsA , tsB ; tsA = parseInt ( ( a . split ( splitMark ) || [ ] ) [ 1 ] || 0 , 36 ) ; tsB = parseInt ( ( b . split ( splitMark ) || [ ] ) [ 1 ] || 0 , 36 ) ; return tsB - tsA ; } ) ; filenamList . length = Math . min ( filenamList . length , this . maxInsetParts ) ; filenamList = filenamList . map ( ( item ) => { return path . join ( this . pathToCargoFolder , item ) ; } ) ; this . uploadLocalUncommitsToClickhouse ( filenamList ) ; }
commit local rotated files to remote ClickHouse DB
commit local rotated files to remote ClickHouse DB
//debuglog "[commitToClickhouseDB]" //if @_isCommiting //debuglog "[commitToClickhouseDB] SKIP is committing" //return //@_isCommiting = true  #lock on //debuglog "[commitToClickhouseDB > readdir] err:", err, ", filenamList:", filenamList //debuglog "[commitToClickhouseDB > readdir] filenamList:", filenamList //@_isCommiting = false  # lock release // filter out non-commits //debuglog "[commitToClickhouseDB] CANCLE empty filenamList" //@_isCommiting = false  # lock release //debuglog "[commitToClickhouseDB > ls] CANCLE empty valid filenamList" //@_isCommiting = false  # lock release // sort filenamList from new to old //debuglog "[commitToClickhouseDB] tsA:#{tsA} tsB:#{tsB}" //debuglog "[commitToClickhouseDB] AFTER SORT filenamList(#{filenamList.length})" #, filenamList // truncate filenamList if too long //debuglog "[commitToClickhouseDB] after truncate filenamList(#{filenamList.length})" #, filenamList // padding full file path // submit each local uncommit sequentially