Merge pull request #22327 from taosdata/fix/TD-25010-3

feat(tsdb/cos): s3 migration

cmake/apr-util_CMakeLists.txt.in

+# apr-util
+ExternalProject_Add(aprutil-1
+        URL https://dlcdn.apache.org//apr/apr-util-1.6.3.tar.gz
+        URL_HASH SHA256=2b74d8932703826862ca305b094eef2983c27b39d5c9414442e9976a9acf1983
+        DOWNLOAD_NO_PROGRESS 1
+        DOWNLOAD_DIR "${TD_CONTRIB_DIR}/deps-download"
+        #GIT_REPOSITORY https://github.com/apache/apr-util.git
+	#GIT_TAG 1.5.4
+        SOURCE_DIR "${TD_CONTRIB_DIR}/apr-util"
+        #BINARY_DIR ""
+        BUILD_IN_SOURCE TRUE
+        BUILD_ALWAYS 1
+        #UPDATE_COMMAND ""
+        CONFIGURE_COMMAND ./configure --prefix=$ENV{HOME}/.cos-local.1/ --with-apr=$ENV{HOME}/.cos-local.1
+        #CONFIGURE_COMMAND ./configure --with-apr=/usr/local/apr
+        BUILD_COMMAND make
+        INSTALL_COMMAND make install
+        TEST_COMMAND ""
+)

cmake/apr_CMakeLists.txt.in

+# apr
+ExternalProject_Add(apr-1
+        URL https://dlcdn.apache.org//apr/apr-1.7.4.tar.gz
+        URL_HASH SHA256=a4137dd82a185076fa50ba54232d920a17c6469c30b0876569e1c2a05ff311d9
+        DOWNLOAD_NO_PROGRESS 1
+        DOWNLOAD_DIR "${TD_CONTRIB_DIR}/deps-download"
+        #GIT_REPOSITORY https://github.com/apache/apr.git
+	#GIT_TAG 1.5.2
+        SOURCE_DIR "${TD_CONTRIB_DIR}/apr"
+        BUILD_IN_SOURCE TRUE
+        UPDATE_DISCONNECTED TRUE
+        BUILD_ALWAYS 1
+        #UPDATE_COMMAND ""
+        CONFIGURE_COMMAND ./configure --prefix=$ENV{HOME}/.cos-local.1/ --enable-shared=no
+        #CONFIGURE_COMMAND ./configure
+        BUILD_COMMAND make
+        INSTALL_COMMAND make install
+        TEST_COMMAND ""
+)

cmake/cmake.options

@@ -125,6 +125,16 @@ option(
     ON
 )
 
+IF(${TD_LINUX})
+
+option(
+    BUILD_WITH_COS
+    "If build with cos"
+    ON
+)
+
+ENDIF ()
+
 option(
     BUILD_WITH_SQLITE
     "If build with sqlite" 

cmake/cos_CMakeLists.txt.in

+# cos
+ExternalProject_Add(cos
+        GIT_REPOSITORY https://github.com/tencentyun/cos-c-sdk-v5.git
+	GIT_TAG v5.0.16
+        SOURCE_DIR "${TD_CONTRIB_DIR}/cos-c-sdk-v5"
+        BINARY_DIR ""
+        #BUILD_IN_SOURCE TRUE
+        CONFIGURE_COMMAND ""
+        BUILD_COMMAND ""
+        INSTALL_COMMAND ""
+        TEST_COMMAND ""
+)

cmake/curl_CMakeLists.txt.in

+# curl
+ExternalProject_Add(curl
+        URL  https://curl.se/download/curl-8.2.1.tar.gz
+        DOWNLOAD_NO_PROGRESS 1
+        DOWNLOAD_DIR "${TD_CONTRIB_DIR}/deps-download"
+        #GIT_REPOSITORY https://github.com/curl/curl.git
+        #GIT_TAG curl-7_88_1
+        SOURCE_DIR "${TD_CONTRIB_DIR}/curl"
+        BUILD_IN_SOURCE TRUE
+        BUILD_ALWAYS 1
+        #UPDATE_COMMAND ""
+        CONFIGURE_COMMAND ./configure --prefix=$ENV{HOME}/.cos-local.1 --without-ssl --enable-shared=no --disable-ldap --disable-ldaps --without-brotli
+        #CONFIGURE_COMMAND ./configure --without-ssl
+        BUILD_COMMAND make
+        INSTALL_COMMAND make install
+        TEST_COMMAND ""
+)

cmake/mxml_CMakeLists.txt.in

+# cos
+ExternalProject_Add(mxml
+        GIT_REPOSITORY https://github.com/michaelrsweet/mxml.git
+	GIT_TAG release-2.10
+        SOURCE_DIR "${TD_CONTRIB_DIR}/mxml"
+        #BINARY_DIR ""
+        BUILD_IN_SOURCE TRUE
+        #UPDATE_COMMAND ""
+        CONFIGURE_COMMAND ./configure --prefix=$ENV{HOME}/.cos-local.1 --enable-shared=no
+        #CONFIGURE_COMMAND ./configure
+        BUILD_COMMAND make
+        INSTALL_COMMAND make install
+        TEST_COMMAND ""
+)

contrib/CMakeLists.txt

@@ -6,6 +6,39 @@ function(cat IN_FILE OUT_FILE)
     file(APPEND ${OUT_FILE} "${CONTENTS}")
 endfunction(cat IN_FILE OUT_FILE)
 
+if(${TD_LINUX})
+
+set(CONTRIB_TMP_FILE3 "${CMAKE_BINARY_DIR}/deps_tmp_CMakeLists.txt.in3")
+configure_file("${TD_SUPPORT_DIR}/deps_CMakeLists.txt.in" ${CONTRIB_TMP_FILE3})
+
+if(${BUILD_WITH_COS})
+    file(MAKE_DIRECTORY $ENV{HOME}/.cos-local.1/)
+    cat("${TD_SUPPORT_DIR}/mxml_CMakeLists.txt.in" ${CONTRIB_TMP_FILE3})
+    cat("${TD_SUPPORT_DIR}/apr_CMakeLists.txt.in" ${CONTRIB_TMP_FILE3})
+    cat("${TD_SUPPORT_DIR}/curl_CMakeLists.txt.in" ${CONTRIB_TMP_FILE3})
+endif(${BUILD_WITH_COS})
+
+configure_file(${CONTRIB_TMP_FILE3} "${TD_CONTRIB_DIR}/deps-download/CMakeLists.txt")
+execute_process(COMMAND "${CMAKE_COMMAND}" -G "${CMAKE_GENERATOR}" .
+        WORKING_DIRECTORY "${TD_CONTRIB_DIR}/deps-download")
+execute_process(COMMAND "${CMAKE_COMMAND}" --build .
+        WORKING_DIRECTORY "${TD_CONTRIB_DIR}/deps-download")
+
+set(CONTRIB_TMP_FILE2 "${CMAKE_BINARY_DIR}/deps_tmp_CMakeLists.txt.in2")
+configure_file("${TD_SUPPORT_DIR}/deps_CMakeLists.txt.in" ${CONTRIB_TMP_FILE2})
+
+if(${BUILD_WITH_COS})
+    cat("${TD_SUPPORT_DIR}/apr-util_CMakeLists.txt.in" ${CONTRIB_TMP_FILE2})
+endif(${BUILD_WITH_COS})
+
+configure_file(${CONTRIB_TMP_FILE2} "${TD_CONTRIB_DIR}/deps-download/CMakeLists.txt")
+execute_process(COMMAND "${CMAKE_COMMAND}" -G "${CMAKE_GENERATOR}" .
+        WORKING_DIRECTORY "${TD_CONTRIB_DIR}/deps-download")
+execute_process(COMMAND "${CMAKE_COMMAND}" --build .
+        WORKING_DIRECTORY "${TD_CONTRIB_DIR}/deps-download")
+
+endif(${TD_LINUX})
+
 set(CONTRIB_TMP_FILE "${CMAKE_BINARY_DIR}/deps_tmp_CMakeLists.txt.in")
 configure_file("${TD_SUPPORT_DIR}/deps_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
 
@@ -122,6 +155,16 @@ if(${BUILD_WITH_SQLITE})
     cat("${TD_SUPPORT_DIR}/sqlite_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
 endif(${BUILD_WITH_SQLITE})
 
+# cos
+if(${BUILD_WITH_COS})
+    #cat("${TD_SUPPORT_DIR}/mxml_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
+    #cat("${TD_SUPPORT_DIR}/apr_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
+    #cat("${TD_SUPPORT_DIR}/apr-util_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
+    #cat("${TD_SUPPORT_DIR}/curl_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
+    cat("${TD_SUPPORT_DIR}/cos_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
+    add_definitions(-DUSE_COS)
+endif(${BUILD_WITH_COS})
+
 # lucene
 if(${BUILD_WITH_LUCENE})
     cat("${TD_SUPPORT_DIR}/lucene_CMakeLists.txt.in" ${CONTRIB_TMP_FILE})
@@ -347,6 +390,31 @@ if (${BUILD_WITH_ROCKSDB})
     endif()
 endif()
 
+# cos
+if(${BUILD_WITH_COS})
+  if(${TD_LINUX})
+    set(CMAKE_PREFIX_PATH $ENV{HOME}/.cos-local.1)
+    #ADD_DEFINITIONS(-DMINIXML_LIBRARY=${CMAKE_BINARY_DIR}/build/lib/libxml.a)
+    option(ENABLE_TEST "Enable the tests" OFF)
+    INCLUDE_DIRECTORIES($ENV{HOME}/.cos-local.1/include)
+    MESSAGE("$ENV{HOME}/.cos-local.1/include")
+
+    set(CMAKE_BUILD_TYPE debug)
+    set(ORIG_CMAKE_PROJECT_NAME ${CMAKE_PROJECT_NAME})
+    set(CMAKE_PROJECT_NAME cos_c_sdk)
+
+    add_subdirectory(cos-c-sdk-v5 EXCLUDE_FROM_ALL)
+    target_include_directories(
+      cos_c_sdk
+      PUBLIC $<BUILD_INTERFACE:${CMAKE_CURRENT_SOURCE_DIR}/cos-c-sdk-v5/cos_c_sdk>
+      )
+
+    set(CMAKE_PROJECT_NAME ${ORIG_CMAKE_PROJECT_NAME})
+  else()
+
+  endif(${TD_LINUX})
+endif(${BUILD_WITH_COS})
+
 # lucene
 # To support build on ubuntu: sudo apt-get install libboost-all-dev
 if(${BUILD_WITH_LUCENE})

contrib/test/CMakeLists.txt

@@ -3,6 +3,11 @@ if(${BUILD_WITH_ROCKSDB})
     add_subdirectory(rocksdb)
 endif(${BUILD_WITH_ROCKSDB})
 
+# cos
+if(${BUILD_WITH_COS})
+    add_subdirectory(cos)
+endif(${BUILD_WITH_COS})
+
 if(${BUILD_WITH_LUCENE})
     add_subdirectory(lucene)
 endif(${BUILD_WITH_LUCENE})

contrib/test/cos/CMakeLists.txt

+add_executable(cosTest "")
+target_sources(cosTest
+  PRIVATE
+  "${CMAKE_CURRENT_SOURCE_DIR}/main.c"
+  )
+
+#find_path(APR_INCLUDE_DIR apr-1/apr_time.h)
+#find_path(APR_UTIL_INCLUDE_DIR apr/include/apr-1/apr_md5.h)
+#find_path(MINIXML_INCLUDE_DIR mxml.h)
+#find_path(CURL_INCLUDE_DIR curl/curl.h)
+
+#include_directories (${MINIXML_INCLUDE_DIR})
+#include_directories (${CURL_INCLUDE_DIR})
+FIND_PROGRAM(APR_CONFIG_BIN NAMES apr-config apr-1-config PATHS /usr/bin /usr/local/bin /usr/local/apr/bin/)
+#FIND_PROGRAM(APU_CONFIG_BIN NAMES apu-config apu-1-config PATHS /usr/bin /usr/local/bin /usr/local/apr/bin/)
+
+IF (APR_CONFIG_BIN)
+  EXECUTE_PROCESS(
+    COMMAND ${APR_CONFIG_BIN} --includedir
+    OUTPUT_VARIABLE APR_INCLUDE_DIR
+    OUTPUT_STRIP_TRAILING_WHITESPACE
+    )
+ENDIF()
+#IF (APU_CONFIG_BIN)
+#  EXECUTE_PROCESS(
+#    COMMAND ${APU_CONFIG_BIN} --includedir
+#    OUTPUT_VARIABLE APR_UTIL_INCLUDE_DIR
+#    OUTPUT_STRIP_TRAILING_WHITESPACE
+#    )
+#ENDIF()
+
+include_directories (${APR_INCLUDE_DIR})
+#include_directories (${APR_UTIL_INCLUDE_DIR})
+
+target_include_directories(
+  cosTest
+  PUBLIC "${TD_SOURCE_DIR}/contrib/cos-c-sdk-v5/cos_c_sdk"
+  )
+
+#find_library(APR_LIBRARY apr-1 PATHS /usr/local/apr/lib/)
+#find_library(APR_UTIL_LIBRARY aprutil-1 PATHS /usr/local/apr/lib/)
+#find_library(MINIXML_LIBRARY mxml)
+#find_library(CURL_LIBRARY curl)
+
+target_link_libraries(cosTest cos_c_sdk)
+target_link_libraries(cosTest apr-1})
+target_link_libraries(cosTest aprutil-1})
+target_link_libraries(cosTest mxml)
+target_link_libraries(cosTest curl)

include/os/osFile.h

@@ -76,7 +76,7 @@ int32_t taosUnLockFile(TdFilePtr pFile);
 
 int32_t taosUmaskFile(int32_t maskVal);
 
-int32_t taosStatFile(const char *path, int64_t *size, int32_t *mtime);
+int32_t taosStatFile(const char *path, int64_t *size, int32_t *mtime, int32_t *atime);
 int32_t taosDevInoFile(TdFilePtr pFile, int64_t *stDev, int64_t *stIno);
 int32_t taosFStatFile(TdFilePtr pFile, int64_t *size, int32_t *mtime);
 bool    taosCheckExistFile(const char *pathname);

source/common/CMakeLists.txt

@@ -16,6 +16,14 @@ ENDIF ()
 IF (TD_STORAGE)
   ADD_DEFINITIONS(-D_STORAGE)
   TARGET_LINK_LIBRARIES(common PRIVATE storage)
+
+  IF(${TD_LINUX})
+    IF(${BUILD_WITH_COS})
+      add_definitions(-DUSE_COS)
+    ENDIF(${BUILD_WITH_COS})
+
+  ENDIF(${TD_LINUX})
+
 ENDIF ()
 
 target_include_directories(

source/common/src/tglobal.c

@@ -237,6 +237,14 @@ int64_t tsCheckpointInterval = 3 * 60 * 60 * 1000;
 bool    tsFilterScalarMode = false;
 int32_t tsKeepTimeOffset = 0;  // latency of data migration
 
+char   tsS3Endpoint[TSDB_FQDN_LEN] = "<endpoint>";
+char   tsS3AccessKey[TSDB_FQDN_LEN] = "<accesskey>";
+char   tsS3AccessKeyId[TSDB_FQDN_LEN] = "<accesskeyid>";
+char   tsS3AccessKeySecret[TSDB_FQDN_LEN] = "<accesskeysecrect>";
+char   tsS3BucketName[TSDB_FQDN_LEN] = "<bucketname>";
+char   tsS3AppId[TSDB_FQDN_LEN] = "<appid>";
+int8_t tsS3Enabled = false;
+
 #ifndef _STORAGE
 int32_t taosSetTfsCfg(SConfig *pCfg) {
   SConfigItem *pItem = cfgGetItem(pCfg, "dataDir");
@@ -258,7 +266,43 @@ int32_t taosSetTfsCfg(SConfig *pCfg) {
 int32_t taosSetTfsCfg(SConfig *pCfg);
 #endif
 
-struct SConfig *taosGetCfg() { return tsCfg; }
+int32_t taosSetS3Cfg(SConfig *pCfg) {
+  tstrncpy(tsS3AccessKey, cfgGetItem(pCfg, "s3Accesskey")->str, TSDB_FQDN_LEN);
+  if (tsS3AccessKey[0] == '<') {
+    return 0;
+  }
+  char *colon = strchr(tsS3AccessKey, ':');
+  if (!colon) {
+    uError("invalid access key:%s", tsS3AccessKey);
+    return -1;
+  }
+  *colon = '\0';
+  tstrncpy(tsS3AccessKeyId, tsS3AccessKey, TSDB_FQDN_LEN);
+  tstrncpy(tsS3AccessKeySecret, colon + 1, TSDB_FQDN_LEN);
+  tstrncpy(tsS3Endpoint, cfgGetItem(pCfg, "s3Endpoint")->str, TSDB_FQDN_LEN);
+  tstrncpy(tsS3BucketName, cfgGetItem(pCfg, "s3BucketName")->str, TSDB_FQDN_LEN);
+  char *cos = strstr(tsS3Endpoint, "cos.");
+  if (cos) {
+    char *appid = strrchr(tsS3BucketName, '-');
+    if (!appid) {
+      uError("failed to locate appid in bucket:%s", tsS3BucketName);
+      return -1;
+    } else {
+      tstrncpy(tsS3AppId, appid + 1, TSDB_FQDN_LEN);
+    }
+  }
+  if (tsS3BucketName[0] != '<' && tsDiskCfgNum > 1) {
+#ifdef USE_COS
+    tsS3Enabled = true;
+#endif
+  }
+
+  return 0;
+}
+
+struct SConfig *taosGetCfg() {
+  return tsCfg;
+}
 
 static int32_t taosLoadCfg(SConfig *pCfg, const char **envCmd, const char *inputCfgDir, const char *envFile,
                            char *apolloUrl) {
@@ -581,6 +625,10 @@ static int32_t taosAddServerCfg(SConfig *pCfg) {
   if (cfgAddInt32(pCfg, "maxStreamBackendCache", tsMaxStreamBackendCache, 16, 1024, CFG_SCOPE_SERVER) != 0) return -1;
   if (cfgAddInt32(pCfg, "pqSortMemThreshold", tsPQSortMemThreshold, 1, 10240, CFG_SCOPE_SERVER) != 0) return -1;
 
+  if (cfgAddString(pCfg, "s3Accesskey", tsS3AccessKey, CFG_SCOPE_SERVER) != 0) return -1;
+  if (cfgAddString(pCfg, "s3Endpoint", tsS3Endpoint, CFG_SCOPE_SERVER) != 0) return -1;
+  if (cfgAddString(pCfg, "s3BucketName", tsS3BucketName, CFG_SCOPE_SERVER) != 0) return -1;
+
   GRANT_CFG_ADD;
   return 0;
 }
@@ -1502,6 +1550,7 @@ int32_t taosInitCfg(const char *cfgDir, const char **envCmd, const char *envFile
     if (taosSetServerCfg(tsCfg)) return -1;
     if (taosSetReleaseCfg(tsCfg)) return -1;
     if (taosSetTfsCfg(tsCfg) != 0) return -1;
+    if (taosSetS3Cfg(tsCfg) != 0) return -1;
   }
   taosSetSystemCfg(tsCfg);
 

source/dnode/mgmt/mgmt_mnode/src/mmFile.c

@@ -46,7 +46,7 @@ static int32_t mmDecodeOption(SJson *pJson, SMnodeOpt *pOption) {
     if (code < 0) return -1;
     tjsonGetInt32ValueFromDouble(replica, "role", pOption->nodeRoles[i], code);
     if (code < 0) return -1;
-    if(pOption->nodeRoles[i] == TAOS_SYNC_ROLE_VOTER){
+    if (pOption->nodeRoles[i] == TAOS_SYNC_ROLE_VOTER) {
       pOption->numOfReplicas++;
     }
   }
@@ -65,7 +65,7 @@ int32_t mmReadFile(const char *path, SMnodeOpt *pOption) {
   char      file[PATH_MAX] = {0};
   snprintf(file, sizeof(file), "%s%smnode.json", path, TD_DIRSEP);
 
-  if (taosStatFile(file, NULL, NULL) < 0) {
+  if (taosStatFile(file, NULL, NULL, NULL) < 0) {
     dInfo("mnode file:%s not exist", file);
     return 0;
   }

source/dnode/mgmt/mgmt_vnode/src/vmFile.c

@@ -97,7 +97,7 @@ int32_t vmGetVnodeListFromFile(SVnodeMgmt *pMgmt, SWrapperCfg **ppCfgs, int32_t
   SWrapperCfg *pCfgs = NULL;
   snprintf(file, sizeof(file), "%s%svnodes.json", pMgmt->path, TD_DIRSEP);
 
-  if (taosStatFile(file, NULL, NULL) < 0) {
+  if (taosStatFile(file, NULL, NULL, NULL) < 0) {
     dInfo("vnode file:%s not exist", file);
     return 0;
   }

source/dnode/mgmt/node_util/src/dmEps.c

@@ -100,7 +100,7 @@ int32_t dmReadEps(SDnodeData *pData) {
     goto _OVER;
   }
 
-  if (taosStatFile(file, NULL, NULL) < 0) {
+  if (taosStatFile(file, NULL, NULL, NULL) < 0) {
     dInfo("dnode file:%s not exist", file);
     code = 0;
     goto _OVER;
@@ -350,7 +350,7 @@ void dmRotateMnodeEpSet(SDnodeData *pData) {
 }
 
 void dmGetMnodeEpSetForRedirect(SDnodeData *pData, SRpcMsg *pMsg, SEpSet *pEpSet) {
-  if(!pData->validMnodeEps) return;
+  if (!pData->validMnodeEps) return;
   dmGetMnodeEpSet(pData, pEpSet);
   dTrace("msg is redirected, handle:%p num:%d use:%d", pMsg->info.handle, pEpSet->numOfEps, pEpSet->inUse);
   for (int32_t i = 0; i < pEpSet->numOfEps; ++i) {
@@ -469,7 +469,7 @@ static int32_t dmReadDnodePairs(SDnodeData *pData) {
   char      file[PATH_MAX] = {0};
   snprintf(file, sizeof(file), "%s%sdnode%sep.json", tsDataDir, TD_DIRSEP, TD_DIRSEP);
 
-  if (taosStatFile(file, NULL, NULL) < 0) {
+  if (taosStatFile(file, NULL, NULL, NULL) < 0) {
     dDebug("dnode file:%s not exist", file);
     code = 0;
     goto _OVER;

source/dnode/mgmt/node_util/src/dmFile.c

@@ -38,7 +38,7 @@ int32_t dmReadFile(const char *path, const char *name, bool *pDeployed) {
   char      file[PATH_MAX] = {0};
   snprintf(file, sizeof(file), "%s%s%s.json", path, TD_DIRSEP, name);
 
-  if (taosStatFile(file, NULL, NULL) < 0) {
+  if (taosStatFile(file, NULL, NULL, NULL) < 0) {
     dInfo("file:%s not exist", file);
     code = 0;
     goto _OVER;

source/dnode/vnode/CMakeLists.txt

@@ -8,6 +8,7 @@ set(
   "src/vnd/vnodeCommit.c"
   "src/vnd/vnodeQuery.c"
   "src/vnd/vnodeModule.c"
+  "src/vnd/vnodeCos.c"
   "src/vnd/vnodeSvr.c"
   "src/vnd/vnodeSync.c"
   "src/vnd/vnodeSnapshot.c"
@@ -155,6 +156,45 @@ target_link_libraries(
     PUBLIC index
 )
 
+if(${TD_LINUX})
+set(CMAKE_FIND_LIBRARY_SUFFIXES ".a")
+find_library(APR_LIBRARY apr-1 PATHS /usr/local/apr/lib/)
+find_library(APR_UTIL_LIBRARY aprutil-1 PATHS /usr/local/apr/lib/)
+find_library(MINIXML_LIBRARY mxml)
+find_library(CURL_LIBRARY curl)
+target_link_libraries(
+    vnode
+
+    # s3
+    PUBLIC cos_c_sdk_static
+    PUBLIC ${APR_UTIL_LIBRARY}
+    PUBLIC ${APR_LIBRARY}
+    PUBLIC ${MINIXML_LIBRARY}
+    PUBLIC ${CURL_LIBRARY}
+)
+
+# s3
+FIND_PROGRAM(APR_CONFIG_BIN NAMES apr-config apr-1-config PATHS /usr/bin /usr/local/bin /usr/local/apr/bin/)
+IF (APR_CONFIG_BIN)
+  EXECUTE_PROCESS(
+    COMMAND ${APR_CONFIG_BIN} --includedir
+    OUTPUT_VARIABLE APR_INCLUDE_DIR
+    OUTPUT_STRIP_TRAILING_WHITESPACE
+    )
+ENDIF()
+include_directories (${APR_INCLUDE_DIR})
+target_include_directories(
+  vnode
+  PUBLIC "${TD_SOURCE_DIR}/contrib/cos-c-sdk-v5/cos_c_sdk"
+  PUBLIC "$ENV{HOME}/.cos-local.1/include"
+  )
+
+if(${BUILD_WITH_COS})
+  add_definitions(-DUSE_COS)
+endif(${BUILD_WITH_COS})
+
+endif(${TD_LINUX})
+
 IF (TD_GRANT)
   TARGET_LINK_LIBRARIES(vnode PUBLIC grant)
 ENDIF ()
@@ -169,8 +209,6 @@ if(${BUILD_WITH_ROCKSDB})
     add_definitions(-DUSE_ROCKSDB)
 endif(${BUILD_WITH_ROCKSDB})
 
-
-
 if(${BUILD_TEST})
     add_subdirectory(test)
 endif(${BUILD_TEST})

source/dnode/vnode/src/inc/vndCos.h

+/*
+ * 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/>.
+ */
+
+#ifndef _TD_VND_COS_H_
+#define _TD_VND_COS_H_
+
+#include "vnd.h"
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+extern int8_t tsS3Enabled;
+
+int32_t s3Init();
+void    s3CleanUp();
+int32_t s3PutObjectFromFile(const char *file, const char *object);
+void    s3DeleteObjects(const char *object_name[], int nobject);
+bool    s3Exists(const char *object_name);
+bool    s3Get(const char *object_name, const char *path);
+void    s3EvictCache(const char *path, long object_size);
+long    s3Size(const char *object_name);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif /*_TD_VND_COS_H_*/

source/dnode/vnode/src/tq/tqOffsetSnapshot.c

@@ -60,7 +60,7 @@ int32_t tqOffsetSnapRead(STqOffsetReader* pReader, uint8_t** ppData) {
   }
 
   int64_t sz = 0;
-  if (taosStatFile(fname, &sz, NULL) < 0) {
+  if (taosStatFile(fname, &sz, NULL, NULL) < 0) {
     taosCloseFile(&pFile);
     taosMemoryFree(fname);
     return -1;

source/dnode/vnode/src/tsdb/tsdbFS.c

@@ -176,7 +176,7 @@ static int32_t tsdbScanAndTryFixFS(STsdb *pTsdb) {
   // SDelFile
   if (pTsdb->fs.pDelFile) {
     tsdbDelFileName(pTsdb, pTsdb->fs.pDelFile, fname);
-    if (taosStatFile(fname, &size, NULL)) {
+    if (taosStatFile(fname, &size, NULL, NULL)) {
       code = TAOS_SYSTEM_ERROR(errno);
       TSDB_CHECK_CODE(code, lino, _exit);
     }
@@ -195,7 +195,7 @@ static int32_t tsdbScanAndTryFixFS(STsdb *pTsdb) {
 
     // head =========
     tsdbHeadFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pHeadF, fname);
-    if (taosStatFile(fname, &size, NULL)) {
+    if (taosStatFile(fname, &size, NULL, NULL)) {
       code = TAOS_SYSTEM_ERROR(errno);
       TSDB_CHECK_CODE(code, lino, _exit);
     }
@@ -206,7 +206,7 @@ static int32_t tsdbScanAndTryFixFS(STsdb *pTsdb) {
 
     // data =========
     tsdbDataFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pDataF, fname);
-    if (taosStatFile(fname, &size, NULL)) {
+    if (taosStatFile(fname, &size, NULL, NULL)) {
       code = TAOS_SYSTEM_ERROR(errno);
       TSDB_CHECK_CODE(code, lino, _exit);
     }
@@ -221,7 +221,7 @@ static int32_t tsdbScanAndTryFixFS(STsdb *pTsdb) {
 
     // sma =============
     tsdbSmaFileName(pTsdb, pSet->diskId, pSet->fid, pSet->pSmaF, fname);
-    if (taosStatFile(fname, &size, NULL)) {
+    if (taosStatFile(fname, &size, NULL, NULL)) {
       code = TAOS_SYSTEM_ERROR(errno);
       TSDB_CHECK_CODE(code, lino, _exit);
     }
@@ -237,7 +237,7 @@ static int32_t tsdbScanAndTryFixFS(STsdb *pTsdb) {
     // stt ===========
     for (int32_t iStt = 0; iStt < pSet->nSttF; iStt++) {
       tsdbSttFileName(pTsdb, pSet->diskId, pSet->fid, pSet->aSttF[iStt], fname);
-      if (taosStatFile(fname, &size, NULL)) {
+      if (taosStatFile(fname, &size, NULL, NULL)) {
         code = TAOS_SYSTEM_ERROR(errno);
         TSDB_CHECK_CODE(code, lino, _exit);
       }

source/dnode/vnode/src/tsdb/tsdbReaderWriter.c

@@ -14,6 +14,7 @@
  */
 
 #include "tsdb.h"
+#include "vndCos.h"
 
 // =============== PAGE-WISE FILE ===============
 int32_t tsdbOpenFile(const char *path, int32_t szPage, int32_t flag, STsdbFD **ppFD) {
@@ -34,9 +35,24 @@ int32_t tsdbOpenFile(const char *path, int32_t szPage, int32_t flag, STsdbFD **p
   pFD->flag = flag;
   pFD->pFD = taosOpenFile(path, flag);
   if (pFD->pFD == NULL) {
-    code = TAOS_SYSTEM_ERROR(errno);
-    taosMemoryFree(pFD);
-    goto _exit;
+    const char *object_name = taosDirEntryBaseName((char *)path);
+    long        s3_size = s3Size(object_name);
+    if (!strncmp(path + strlen(path) - 5, ".data", 5) && s3_size > 0) {
+      s3EvictCache(path, s3_size);
+      s3Get(object_name, path);
+
+      pFD->pFD = taosOpenFile(path, flag);
+
+      if (pFD->pFD == NULL) {
+        code = TAOS_SYSTEM_ERROR(errno);
+        taosMemoryFree(pFD);
+        goto _exit;
+      }
+    } else {
+      code = TAOS_SYSTEM_ERROR(errno);
+      taosMemoryFree(pFD);
+      goto _exit;
+    }
   }
   pFD->szPage = szPage;
   pFD->pgno = 0;
@@ -50,7 +66,7 @@ int32_t tsdbOpenFile(const char *path, int32_t szPage, int32_t flag, STsdbFD **p
 
   // not check file size when reading data files.
   if (flag != TD_FILE_READ) {
-    if (taosStatFile(path, &pFD->szFile, NULL) < 0) {
+    if (taosStatFile(path, &pFD->szFile, NULL, NULL) < 0) {
       code = TAOS_SYSTEM_ERROR(errno);
       taosMemoryFree(pFD->pBuf);
       taosCloseFile(&pFD->pFD);

source/dnode/vnode/src/tsdb/tsdbRetention.c

@@ -15,6 +15,7 @@
 
 #include "tsdb.h"
 #include "tsdbFS2.h"
+#include "vndCos.h"
 
 typedef struct {
   STsdb  *tsdb;
@@ -41,6 +42,28 @@ static int32_t tsdbDoRemoveFileObject(SRTNer *rtner, const STFileObj *fobj) {
   return TARRAY2_APPEND(rtner->fopArr, op);
 }
 
+static int32_t tsdbRemoveFileObjectS3(SRTNer *rtner, const STFileObj *fobj) {
+  int32_t code = 0, lino = 0;
+
+  STFileOp op = {
+      .optype = TSDB_FOP_REMOVE,
+      .fid = fobj->f->fid,
+      .of = fobj->f[0],
+  };
+
+  code = TARRAY2_APPEND(rtner->fopArr, op);
+  TSDB_CHECK_CODE(code, lino, _exit);
+
+  const char *object_name = taosDirEntryBaseName((char *)fobj->fname);
+  s3DeleteObjects(&object_name, 1);
+
+_exit:
+  if (code) {
+    TSDB_ERROR_LOG(TD_VID(rtner->tsdb->pVnode), lino, code);
+  }
+  return code;
+}
+
 static int32_t tsdbDoCopyFile(SRTNer *rtner, const STFileObj *from, const STFile *to) {
   int32_t code = 0;
   int32_t lino = 0;
@@ -76,6 +99,34 @@ _exit:
   return code;
 }
 
+static int32_t tsdbCopyFileS3(SRTNer *rtner, const STFileObj *from, const STFile *to) {
+  int32_t code = 0;
+  int32_t lino = 0;
+
+  char      fname[TSDB_FILENAME_LEN];
+  TdFilePtr fdFrom = NULL;
+  TdFilePtr fdTo = NULL;
+
+  tsdbTFileName(rtner->tsdb, to, fname);
+
+  fdFrom = taosOpenFile(from->fname, TD_FILE_READ);
+  if (fdFrom == NULL) code = terrno;
+  TSDB_CHECK_CODE(code, lino, _exit);
+
+  char *object_name = taosDirEntryBaseName(fname);
+  code = s3PutObjectFromFile(from->fname, object_name);
+  TSDB_CHECK_CODE(code, lino, _exit);
+
+  taosCloseFile(&fdFrom);
+
+_exit:
+  if (code) {
+    TSDB_ERROR_LOG(TD_VID(rtner->tsdb->pVnode), lino, code);
+    taosCloseFile(&fdFrom);
+  }
+  return code;
+}
+
 static int32_t tsdbDoMigrateFileObj(SRTNer *rtner, const STFileObj *fobj, const SDiskID *did) {
   int32_t  code = 0;
   int32_t  lino = 0;
@@ -123,6 +174,53 @@ _exit:
   return code;
 }
 
+static int32_t tsdbMigrateDataFileS3(SRTNer *rtner, const STFileObj *fobj, const SDiskID *did) {
+  int32_t  code = 0;
+  int32_t  lino = 0;
+  STFileOp op = {0};
+
+  // remove old
+  op = (STFileOp){
+      .optype = TSDB_FOP_REMOVE,
+      .fid = fobj->f->fid,
+      .of = fobj->f[0],
+  };
+
+  code = TARRAY2_APPEND(rtner->fopArr, op);
+  TSDB_CHECK_CODE(code, lino, _exit);
+
+  // create new
+  op = (STFileOp){
+      .optype = TSDB_FOP_CREATE,
+      .fid = fobj->f->fid,
+      .nf =
+          {
+              .type = fobj->f->type,
+              .did = did[0],
+              .fid = fobj->f->fid,
+              .cid = fobj->f->cid,
+              .size = fobj->f->size,
+              .stt[0] =
+                  {
+                      .level = fobj->f->stt[0].level,
+                  },
+          },
+  };
+
+  code = TARRAY2_APPEND(rtner->fopArr, op);
+  TSDB_CHECK_CODE(code, lino, _exit);
+
+  // do copy the file
+  code = tsdbCopyFileS3(rtner, fobj, &op.nf);
+  TSDB_CHECK_CODE(code, lino, _exit);
+
+_exit:
+  if (code) {
+    TSDB_ERROR_LOG(TD_VID(rtner->tsdb->pVnode), lino, code);
+  }
+  return code;
+}
+
 typedef struct {
   STsdb  *tsdb;
   int32_t sync;
@@ -201,8 +299,14 @@ static int32_t tsdbDoRetention2(void *arg) {
       for (int32_t ftype = 0; (ftype < TSDB_FTYPE_MAX) && (fobj = rtner->ctx->fset->farr[ftype], 1); ++ftype) {
         if (fobj == NULL) continue;
 
-        code = tsdbDoRemoveFileObject(rtner, fobj);
-        TSDB_CHECK_CODE(code, lino, _exit);
+        int32_t nlevel = tfsGetLevel(rtner->tsdb->pVnode->pTfs);
+        if (tsS3Enabled && nlevel > 1 && TSDB_FTYPE_DATA == ftype && fobj->f->did.level == nlevel - 1) {
+          code = tsdbRemoveFileObjectS3(rtner, fobj);
+          TSDB_CHECK_CODE(code, lino, _exit);
+        } else {
+          code = tsdbDoRemoveFileObject(rtner, fobj);
+          TSDB_CHECK_CODE(code, lino, _exit);
+        }
       }
 
       SSttLvl *lvl;
@@ -228,8 +332,15 @@ static int32_t tsdbDoRetention2(void *arg) {
         if (fobj == NULL) continue;
 
         if (fobj->f->did.level == did.level) continue;
-        code = tsdbDoMigrateFileObj(rtner, fobj, &did);
-        TSDB_CHECK_CODE(code, lino, _exit);
+
+        int32_t nlevel = tfsGetLevel(rtner->tsdb->pVnode->pTfs);
+        if (tsS3Enabled && nlevel > 1 && TSDB_FTYPE_DATA == ftype && did.level == nlevel - 1) {
+          code = tsdbMigrateDataFileS3(rtner, fobj, &did);
+          TSDB_CHECK_CODE(code, lino, _exit);
+        } else {
+          code = tsdbDoMigrateFileObj(rtner, fobj, &did);
+          TSDB_CHECK_CODE(code, lino, _exit);
+        }
       }
 
       // stt
@@ -281,4 +392,4 @@ int32_t tsdbRetention(STsdb *tsdb, int64_t now, int32_t sync) {
     tsdbFreeRtnArg(arg);
   }
   return code;
-}
+}
\ No newline at end of file

source/dnode/vnode/src/tsdb/tsdbWrite.c

@@ -76,7 +76,7 @@ int tsdbScanAndConvertSubmitMsg(STsdb *pTsdb, SSubmitReq2 *pMsg) {
   int32_t       code = 0;
   STsdbKeepCfg *pCfg = &pTsdb->keepCfg;
   TSKEY         now = taosGetTimestamp(pCfg->precision);
-  TSKEY         minKey = now - tsTickPerMin[pCfg->precision] * pCfg->keep2;
+  TSKEY         minKey = now - tsTickPerMin[pCfg->precision] * pCfg->keep1;
   TSKEY         maxKey = tsMaxKeyByPrecision[pCfg->precision];
   int32_t       size = taosArrayGetSize(pMsg->aSubmitTbData);
 
@@ -107,4 +107,4 @@ int tsdbScanAndConvertSubmitMsg(STsdb *pTsdb, SSubmitReq2 *pMsg) {
 
 _exit:
   return code;
-}
+}
\ No newline at end of file

source/dnode/vnode/src/vnd/vnodeCos.c

+#define ALLOW_FORBID_FUNC
+
+#include "vndCos.h"
+
+extern char tsS3Endpoint[];
+extern char tsS3AccessKeyId[];
+extern char tsS3AccessKeySecret[];
+extern char tsS3BucketName[];
+extern char tsS3AppId[];
+
+#ifdef USE_COS
+#include "cos_api.h"
+#include "cos_http_io.h"
+#include "cos_log.h"
+
+int32_t s3Init() {
+  if (cos_http_io_initialize(NULL, 0) != COSE_OK) {
+    return -1;
+  }
+
+  // set log level, default COS_LOG_WARN
+  cos_log_set_level(COS_LOG_WARN);
+
+  // set log output, default stderr
+  cos_log_set_output(NULL);
+
+  return 0;
+}
+
+void s3CleanUp() { cos_http_io_deinitialize(); }
+
+static void log_status(cos_status_t *s) {
+  cos_warn_log("status->code: %d", s->code);
+  if (s->error_code) cos_warn_log("status->error_code: %s", s->error_code);
+  if (s->error_msg) cos_warn_log("status->error_msg: %s", s->error_msg);
+  if (s->req_id) cos_warn_log("status->req_id: %s", s->req_id);
+}
+
+static void s3InitRequestOptions(cos_request_options_t *options, int is_cname) {
+  options->config = cos_config_create(options->pool);
+
+  cos_config_t *config = options->config;
+
+  cos_str_set(&config->endpoint, tsS3Endpoint);
+  cos_str_set(&config->access_key_id, tsS3AccessKeyId);
+  cos_str_set(&config->access_key_secret, tsS3AccessKeySecret);
+  cos_str_set(&config->appid, tsS3AppId);
+
+  config->is_cname = is_cname;
+
+  options->ctl = cos_http_controller_create(options->pool, 0);
+}
+
+int32_t s3PutObjectFromFile(const char *file_str, const char *object_str) {
+  int32_t                code = 0;
+  cos_pool_t            *p = NULL;
+  int                    is_cname = 0;
+  cos_status_t          *s = NULL;
+  cos_request_options_t *options = NULL;
+  cos_string_t           bucket, object, file;
+  cos_table_t           *resp_headers;
+  int                    traffic_limit = 0;
+
+  cos_pool_create(&p, NULL);
+  options = cos_request_options_create(p);
+  s3InitRequestOptions(options, is_cname);
+  cos_table_t *headers = NULL;
+  if (traffic_limit) {
+    // 限速值设置范围为819200 - 838860800，即100KB/s - 100MB/s，如果超出该范围将返回400错误
+    headers = cos_table_make(p, 1);
+    cos_table_add_int(headers, "x-cos-traffic-limit", 819200);
+  }
+  cos_str_set(&bucket, tsS3BucketName);
+  cos_str_set(&file, file_str);
+  cos_str_set(&object, object_str);
+  s = cos_put_object_from_file(options, &bucket, &object, &file, headers, &resp_headers);
+  log_status(s);
+
+  cos_pool_destroy(p);
+
+  if (s->code != 200) {
+    return code = s->code;
+  }
+
+  return code;
+}
+
+void s3DeleteObjects(const char *object_name[], int nobject) {
+  cos_pool_t            *p = NULL;
+  int                    is_cname = 0;
+  cos_string_t           bucket;
+  cos_table_t           *resp_headers = NULL;
+  cos_request_options_t *options = NULL;
+  cos_list_t             object_list;
+  cos_list_t             deleted_object_list;
+  int                    is_quiet = COS_TRUE;
+
+  cos_pool_create(&p, NULL);
+  options = cos_request_options_create(p);
+  s3InitRequestOptions(options, is_cname);
+  cos_str_set(&bucket, tsS3BucketName);
+
+  cos_list_init(&object_list);
+  cos_list_init(&deleted_object_list);
+
+  for (int i = 0; i < nobject; ++i) {
+    cos_object_key_t *content = cos_create_cos_object_key(p);
+    cos_str_set(&content->key, object_name[i]);
+    cos_list_add_tail(&content->node, &object_list);
+  }
+
+  cos_status_t *s = cos_delete_objects(options, &bucket, &object_list, is_quiet, &resp_headers, &deleted_object_list);
+  log_status(s);
+
+  cos_pool_destroy(p);
+
+  if (cos_status_is_ok(s)) {
+    cos_warn_log("delete objects succeeded\n");
+  } else {
+    cos_warn_log("delete objects failed\n");
+  }
+}
+
+bool s3Exists(const char *object_name) {
+  bool                      ret = false;
+  cos_pool_t               *p = NULL;
+  int                       is_cname = 0;
+  cos_status_t             *s = NULL;
+  cos_request_options_t    *options = NULL;
+  cos_string_t              bucket;
+  cos_string_t              object;
+  cos_table_t              *resp_headers;
+  cos_table_t              *headers = NULL;
+  cos_object_exist_status_e object_exist;
+
+  cos_pool_create(&p, NULL);
+  options = cos_request_options_create(p);
+  s3InitRequestOptions(options, is_cname);
+  cos_str_set(&bucket, tsS3BucketName);
+  cos_str_set(&object, object_name);
+
+  s = cos_check_object_exist(options, &bucket, &object, headers, &object_exist, &resp_headers);
+  if (object_exist == COS_OBJECT_NON_EXIST) {
+    cos_warn_log("object: %.*s non exist.\n", object.len, object.data);
+  } else if (object_exist == COS_OBJECT_EXIST) {
+    ret = true;
+    cos_warn_log("object: %.*s exist.\n", object.len, object.data);
+  } else {
+    cos_warn_log("object: %.*s unknown status.\n", object.len, object.data);
+    log_status(s);
+  }
+
+  cos_pool_destroy(p);
+
+  return ret;
+}
+
+bool s3Get(const char *object_name, const char *path) {
+  bool                   ret = false;
+  cos_pool_t            *p = NULL;
+  int                    is_cname = 0;
+  cos_status_t          *s = NULL;
+  cos_request_options_t *options = NULL;
+  cos_string_t           bucket;
+  cos_string_t           object;
+  cos_string_t           file;
+  cos_table_t           *resp_headers = NULL;
+  cos_table_t           *headers = NULL;
+  int                    traffic_limit = 0;
+
+  //创建内存池
+  cos_pool_create(&p, NULL);
+
+  //初始化请求选项
+  options = cos_request_options_create(p);
+  s3InitRequestOptions(options, is_cname);
+  cos_str_set(&bucket, tsS3BucketName);
+  if (traffic_limit) {
+    //限速值设置范围为819200 - 838860800，即100KB/s - 100MB/s，如果超出该范围将返回400错误
+    headers = cos_table_make(p, 1);
+    cos_table_add_int(headers, "x-cos-traffic-limit", 819200);
+  }
+
+  //下载对象
+  cos_str_set(&file, path);
+  cos_str_set(&object, object_name);
+  s = cos_get_object_to_file(options, &bucket, &object, headers, NULL, &file, &resp_headers);
+  if (cos_status_is_ok(s)) {
+    ret = true;
+    cos_warn_log("get object succeeded\n");
+  } else {
+    cos_warn_log("get object failed\n");
+  }
+
+  //销毁内存池
+  cos_pool_destroy(p);
+
+  return ret;
+}
+
+typedef struct {
+  int64_t size;
+  int32_t atime;
+  char    name[TSDB_FILENAME_LEN];
+} SEvictFile;
+
+static int32_t evictFileCompareAsce(const void *pLeft, const void *pRight) {
+  SEvictFile *lhs = (SEvictFile *)pLeft;
+  SEvictFile *rhs = (SEvictFile *)pRight;
+  return lhs->atime < rhs->atime ? -1 : 1;
+}
+
+void s3EvictCache(const char *path, long object_size) {
+  SDiskSize disk_size = {0};
+  char      dir_name[TSDB_FILENAME_LEN] = "\0";
+
+  tstrncpy(dir_name, path, TSDB_FILENAME_LEN);
+  taosDirName(dir_name);
+
+  if (taosGetDiskSize((char *)dir_name, &disk_size) < 0) {
+    terrno = TAOS_SYSTEM_ERROR(errno);
+    vError("failed to get disk:%s size since %s", path, terrstr());
+    return;
+  }
+
+  if (object_size >= disk_size.avail - (1 << 30)) {
+    // evict too old files
+    // 1, list data files' atime under dir(path)
+    tdbDirPtr pDir = taosOpenDir(dir_name);
+    if (pDir == NULL) {
+      terrno = TAOS_SYSTEM_ERROR(errno);
+      vError("failed to open %s since %s", dir_name, terrstr());
+    }
+    SArray        *evict_files = taosArrayInit(16, sizeof(SEvictFile));
+    tdbDirEntryPtr pDirEntry;
+    while ((pDirEntry = taosReadDir(pDir)) != NULL) {
+      char *name = taosGetDirEntryName(pDirEntry);
+      if (!strncmp(name + strlen(name) - 5, ".data", 5)) {
+        SEvictFile e_file = {0};
+        char       entry_name[TSDB_FILENAME_LEN] = "\0";
+        int        dir_len = strlen(dir_name);
+
+        memcpy(e_file.name, dir_name, dir_len);
+        e_file.name[dir_len] = '/';
+        memcpy(e_file.name + dir_len + 1, name, strlen(name));
+
+        taosStatFile(e_file.name, &e_file.size, NULL, &e_file.atime);
+
+        taosArrayPush(evict_files, &e_file);
+      }
+    }
+    taosCloseDir(&pDir);
+
+    // 2, sort by atime
+    taosArraySort(evict_files, evictFileCompareAsce);
+
+    // 3, remove files ascendingly until we get enough object_size space
+    long   evict_size = 0;
+    size_t ef_size = TARRAY_SIZE(evict_files);
+    for (size_t i = 0; i < ef_size; ++i) {
+      SEvictFile *evict_file = taosArrayGet(evict_files, i);
+      taosRemoveFile(evict_file->name);
+      evict_size += evict_file->size;
+      if (evict_size >= object_size) {
+        break;
+      }
+    }
+
+    taosArrayDestroy(evict_files);
+  }
+}
+
+long s3Size(const char *object_name) {
+  long size = 0;
+
+  cos_pool_t            *p = NULL;
+  int                    is_cname = 0;
+  cos_status_t          *s = NULL;
+  cos_request_options_t *options = NULL;
+  cos_string_t           bucket;
+  cos_string_t           object;
+  cos_table_t           *resp_headers = NULL;
+
+  //创建内存池
+  cos_pool_create(&p, NULL);
+
+  //初始化请求选项
+  options = cos_request_options_create(p);
+  s3InitRequestOptions(options, is_cname);
+  cos_str_set(&bucket, tsS3BucketName);
+
+  //获取对象元数据
+  cos_str_set(&object, object_name);
+  s = cos_head_object(options, &bucket, &object, NULL, &resp_headers);
+  // print_headers(resp_headers);
+  if (cos_status_is_ok(s)) {
+    char *content_length_str = (char *)apr_table_get(resp_headers, COS_CONTENT_LENGTH);
+    if (content_length_str != NULL) {
+      size = atol(content_length_str);
+    }
+    cos_warn_log("head object succeeded: %ld\n", size);
+  } else {
+    cos_warn_log("head object failed\n");
+  }
+
+  //销毁内存池
+  cos_pool_destroy(p);
+
+  return size;
+}
+
+#else
+
+int32_t s3Init() { return 0; }
+void    s3CleanUp() {}
+int32_t s3PutObjectFromFile(const char *file, const char *object) { return 0; }
+void    s3DeleteObjects(const char *object_name[], int nobject) {}
+bool    s3Exists(const char *object_name) { return false; }
+bool    s3Get(const char *object_name, const char *path) { return false; }
+void    s3EvictCache(const char *path, long object_size) {}
+long    s3Size(const char *object_name) { return 0; }
+
+#endif

source/dnode/vnode/src/vnd/vnodeModule.c

@@ -14,6 +14,7 @@
  */
 
 #include "vnd.h"
+#include "vndCos.h"
 
 typedef struct SVnodeTask SVnodeTask;
 struct SVnodeTask {
@@ -81,6 +82,9 @@ int vnodeInit(int nthreads) {
   if (tqInit() < 0) {
     return -1;
   }
+  if (s3Init() < 0) {
+    return -1;
+  }
 
   return 0;
 }
@@ -112,6 +116,7 @@ void vnodeCleanup() {
   walCleanUp();
   tqCleanUp();
   smaCleanUp();
+  s3CleanUp();
 }
 
 int vnodeScheduleTaskEx(int tpid, int (*execute)(void*), void* arg) {

source/libs/function/src/udfd.c

@@ -378,9 +378,9 @@ int32_t udfdInitializePythonPlugin(SUdfScriptPlugin *plugin) {
                                                  "pyUdfDestroy",   "pyUdfScalarProc", "pyUdfAggStart",
                                                  "pyUdfAggFinish", "pyUdfAggProc",    "pyUdfAggMerge"};
   void      **funcs[UDFD_MAX_PLUGIN_FUNCS] = {
-      (void **)&plugin->openFunc,         (void **)&plugin->closeFunc,         (void **)&plugin->udfInitFunc,
-      (void **)&plugin->udfDestroyFunc,   (void **)&plugin->udfScalarProcFunc, (void **)&plugin->udfAggStartFunc,
-      (void **)&plugin->udfAggFinishFunc, (void **)&plugin->udfAggProcFunc,    (void **)&plugin->udfAggMergeFunc};
+           (void **)&plugin->openFunc,         (void **)&plugin->closeFunc,         (void **)&plugin->udfInitFunc,
+           (void **)&plugin->udfDestroyFunc,   (void **)&plugin->udfScalarProcFunc, (void **)&plugin->udfAggStartFunc,
+           (void **)&plugin->udfAggFinishFunc, (void **)&plugin->udfAggProcFunc,    (void **)&plugin->udfAggMergeFunc};
   int32_t err = udfdLoadSharedLib(plugin->libPath, &plugin->lib, funcName, funcs, UDFD_MAX_PLUGIN_FUNCS);
   if (err != 0) {
     fnError("can not load python plugin. lib path %s", plugin->libPath);
@@ -848,7 +848,7 @@ int32_t udfdSaveFuncBodyToFile(SFuncInfo *pFuncInfo, SUdf *udf) {
 
   char path[PATH_MAX] = {0};
   udfdGetFuncBodyPath(udf, path);
-  bool fileExist = !(taosStatFile(path, NULL, NULL) < 0);
+  bool fileExist = !(taosStatFile(path, NULL, NULL, NULL) < 0);
   if (fileExist) {
     strncpy(udf->path, path, PATH_MAX);
     fnInfo("udfd func body file. reuse existing file %s", path);

source/libs/index/src/indexFstFile.c

@@ -162,7 +162,7 @@ static FORCE_INLINE int idxFileCtxGetSize(IFileCtx* ctx) {
       return ctx->offset;
     } else {
       int64_t file_size = 0;
-      taosStatFile(ctx->file.buf, &file_size, NULL);
+      taosStatFile(ctx->file.buf, &file_size, NULL, NULL);
       return (int)file_size;
     }
   }
@@ -199,7 +199,7 @@ IFileCtx* idxFileCtxCreate(WriterType type, const char* path, bool readOnly, int
       code = taosFtruncateFile(ctx->file.pFile, 0);
       UNUSED(code);
 
-      code = taosStatFile(path, &ctx->file.size, NULL);
+      code = taosStatFile(path, &ctx->file.size, NULL, NULL);
       UNUSED(code);
 
       ctx->file.wBufOffset = 0;

source/libs/sync/src/syncRaftStore.c

@@ -42,7 +42,7 @@ int32_t raftStoreReadFile(SSyncNode *pNode) {
   const char *file = pNode->raftStorePath;
   SRaftStore *pStore = &pNode->raftStore;
 
-  if (taosStatFile(file, NULL, NULL) < 0) {
+  if (taosStatFile(file, NULL, NULL, NULL) < 0) {
     sInfo("vgId:%d, raft store file:%s not exist, use default value", pNode->vgId, file);
     pStore->currentTerm = 0;
     pStore->voteFor.addr = 0;

source/libs/wal/src/walMeta.c

@@ -53,7 +53,7 @@ static FORCE_INLINE int64_t walScanLogGetLastVer(SWal* pWal, int32_t fileIdx) {
   walBuildLogName(pWal, pFileInfo->firstVer, fnameStr);
 
   int64_t fileSize = 0;
-  taosStatFile(fnameStr, &fileSize, NULL);
+  taosStatFile(fnameStr, &fileSize, NULL, NULL);
 
   TdFilePtr pFile = taosOpenFile(fnameStr, TD_FILE_READ | TD_FILE_WRITE);
   if (pFile == NULL) {
@@ -304,7 +304,7 @@ int walRepairLogFileTs(SWal* pWal, bool* updateMeta) {
 
     walBuildLogName(pWal, pFileInfo->firstVer, fnameStr);
     int32_t mtime = 0;
-    if (taosStatFile(fnameStr, NULL, &mtime) < 0) {
+    if (taosStatFile(fnameStr, NULL, &mtime, NULL) < 0) {
       terrno = TAOS_SYSTEM_ERROR(errno);
       wError("vgId:%d, failed to stat file due to %s, file:%s", pWal->cfg.vgId, strerror(errno), fnameStr);
       return -1;
@@ -353,7 +353,7 @@ int walTrimIdxFile(SWal* pWal, int32_t fileIdx) {
   walBuildIdxName(pWal, pFileInfo->firstVer, fnameStr);
 
   int64_t fileSize = 0;
-  taosStatFile(fnameStr, &fileSize, NULL);
+  taosStatFile(fnameStr, &fileSize, NULL, NULL);
   int64_t records = TMAX(0, pFileInfo->lastVer - pFileInfo->firstVer + 1);
   int64_t lastEndOffset = records * sizeof(SWalIdxEntry);
 
@@ -436,7 +436,7 @@ int walCheckAndRepairMeta(SWal* pWal) {
     SWalFileInfo* pFileInfo = taosArrayGet(pWal->fileInfoSet, fileIdx);
 
     walBuildLogName(pWal, pFileInfo->firstVer, fnameStr);
-    int32_t code = taosStatFile(fnameStr, &fileSize, NULL);
+    int32_t code = taosStatFile(fnameStr, &fileSize, NULL, NULL);
     if (code < 0) {
       terrno = TAOS_SYSTEM_ERROR(errno);
       wError("failed to stat file since %s. file:%s", terrstr(), fnameStr);
@@ -522,7 +522,7 @@ int walCheckAndRepairIdxFile(SWal* pWal, int32_t fileIdx) {
   walBuildLogName(pWal, pFileInfo->firstVer, fLogNameStr);
   int64_t fileSize = 0;
 
-  if (taosStatFile(fnameStr, &fileSize, NULL) < 0 && errno != ENOENT) {
+  if (taosStatFile(fnameStr, &fileSize, NULL, NULL) < 0 && errno != ENOENT) {
     wError("vgId:%d, failed to stat file due to %s. file:%s", pWal->cfg.vgId, strerror(errno), fnameStr);
     terrno = TAOS_SYSTEM_ERROR(errno);
     return -1;
@@ -935,7 +935,7 @@ int walLoadMeta(SWal* pWal) {
   walBuildMetaName(pWal, metaVer, fnameStr);
   // read metafile
   int64_t fileSize = 0;
-  taosStatFile(fnameStr, &fileSize, NULL);
+  taosStatFile(fnameStr, &fileSize, NULL, NULL);
   if (fileSize == 0) {
     (void)taosRemoveFile(fnameStr);
     wDebug("vgId:%d, wal find empty meta ver %d", pWal->cfg.vgId, metaVer);

source/os/src/osFile.c

@@ -191,7 +191,7 @@ int32_t taosRenameFile(const char *oldName, const char *newName) {
 #endif
 }
 
-int32_t taosStatFile(const char *path, int64_t *size, int32_t *mtime) {
+int32_t taosStatFile(const char *path, int64_t *size, int32_t *mtime, int32_t *atime) {
 #ifdef WINDOWS
   struct _stati64 fileStat;
   int32_t         code = _stati64(path, &fileStat);
@@ -211,6 +211,10 @@ int32_t taosStatFile(const char *path, int64_t *size, int32_t *mtime) {
     *mtime = fileStat.st_mtime;
   }
 
+  if (atime != NULL) {
+    *atime = fileStat.st_mtime;
+  }
+
   return 0;
 }
 int32_t taosDevInoFile(TdFilePtr pFile, int64_t *stDev, int64_t *stIno) {
@@ -540,7 +544,7 @@ int32_t taosFStatFile(TdFilePtr pFile, int64_t *size, int32_t *mtime) {
 
 #ifdef WINDOWS
   struct __stat64 fileStat;
-  int32_t code = _fstat64(pFile->fd, &fileStat);
+  int32_t         code = _fstat64(pFile->fd, &fileStat);
 #else
   struct stat fileStat;
   int32_t code = fstat(pFile->fd, &fileStat);
@@ -897,17 +901,17 @@ int32_t taosCompressFile(char *srcFileName, char *destFileName) {
     goto cmp_end;
   }
 
-   dstFp = gzdopen(pFile->fd, "wb6f");
-   if (dstFp == NULL) {
-     ret = -3;
-     taosCloseFile(&pFile);
-     goto cmp_end;
-   }
+  dstFp = gzdopen(pFile->fd, "wb6f");
+  if (dstFp == NULL) {
+    ret = -3;
+    taosCloseFile(&pFile);
+    goto cmp_end;
+  }
 
-   while (!feof(pSrcFile->fp)) {
-     len = (int32_t)fread(data, 1, compressSize, pSrcFile->fp);
-     (void)gzwrite(dstFp, data, len);
-   }
+  while (!feof(pSrcFile->fp)) {
+    len = (int32_t)fread(data, 1, compressSize, pSrcFile->fp);
+    (void)gzwrite(dstFp, data, len);
+  }
 
 cmp_end:
   if (pSrcFile) {

tests/parallel_test/container_build.sh

@@ -60,6 +60,7 @@ docker run \
     -v /root/.cargo/git:/root/.cargo/git \
     -v /root/go/pkg/mod:/root/go/pkg/mod \
     -v /root/.cache/go-build:/root/.cache/go-build \
+    -v /root/.cos-local.1:/root/.cos-local.1 \
     -v ${REP_REAL_PATH}/enterprise/src/plugins/taosx/target:${REP_DIR}/enterprise/src/plugins/taosx/target \
     -v ${REP_REAL_PATH}/community/tools/taosws-rs/target:${REP_DIR}/community/tools/taosws-rs/target \
     -v ${REP_REAL_PATH}/community/contrib/cJson/:${REP_DIR}/community/contrib/cJson \
@@ -88,6 +89,7 @@ docker run \
     -v /root/.cargo/git:/root/.cargo/git \
     -v /root/go/pkg/mod:/root/go/pkg/mod \
     -v /root/.cache/go-build:/root/.cache/go-build \
+    -v /root/.cos-local.1:/root/.cos-local.1 \
     -v ${REP_REAL_PATH}/enterprise/src/plugins/taosx/target:${REP_DIR}/enterprise/src/plugins/taosx/target \
     -v ${REP_REAL_PATH}/community/tools/taosws-rs/target:${REP_DIR}/community/tools/taosws-rs/target \
     -v ${REP_REAL_PATH}/community/contrib/cJson/:${REP_DIR}/community/contrib/cJson \

utils/test/c/tmqDemo.c

@@ -221,7 +221,7 @@ int64_t getDirectorySize(char* dir) {
       totalSize += subDirSize;
     } else if (0 == strcmp(strchr(fileName, '.'), ".log")) {  // only calc .log file size, and not include .idx file
       int64_t file_size = 0;
-      taosStatFile(subdir, &file_size, NULL);
+      taosStatFile(subdir, &file_size, NULL, NULL);
       totalSize += file_size;
     }
   }
@@ -702,4 +702,3 @@ int main(int32_t argc, char* argv[]) {
   taosCloseFile(&g_fp);
   return 0;
 }
-