Merge branch 'branch-0.3.1-yuncong' into branch-0.3.1

Former-commit-id: 92f799018746a7b117aa0ec1e4af15b0d13a4c01

CHANGELOGS.md

@@ -17,3 +17,4 @@ Please mark all change in change log and use the ticket from JIRA.
 - MS-1 - Add CHANGELOG.md
 - MS-161 - Add CI / CD Module to Milvus Project
 - MS-202 - Add Milvus Jenkins project email notification
+- MS-215 - Add Milvus cluster CI/CD groovy file 

ci/jenkinsfile/cluster_deploy2dev.groovy

@@ -3,13 +3,15 @@ try {
     sh 'helm repo add milvus https://registry.zilliz.com/chartrepo/milvus'
     sh 'helm repo update'
     dir ("milvus-helm") {
-        checkout([$class: 'GitSCM', branches: [[name: "${SEMVER}"]], doGenerateSubmoduleConfigurations: false, extensions: [[$class: 'SubmoduleOption',disableSubmodules: false,parentCredentials: true,recursiveSubmodules: true,reference: '',trackingSubmodules: false]], submoduleCfg: [], userRemoteConfigs: [[credentialsId: "${params.GIT_USER}", url: "git@192.168.1.105:megasearch/milvus-helm.git", name: 'origin', refspec: "+refs/heads/${SEMVER}:refs/remotes/origin/${SEMVER}"]]])
+        checkout([$class: 'GitSCM', branches: [[name: "${SEMVER}"]], doGenerateSubmoduleConfigurations: false, extensions: [], submoduleCfg: [], userRemoteConfigs: [[credentialsId: "${params.GIT_USER}", url: "git@192.168.1.105:megasearch/milvus-helm.git", name: 'origin', refspec: "+refs/heads/${SEMVER}:refs/remotes/origin/${SEMVER}"]]])
         dir ("milvus/milvus-cluster") {
-            sh "helm install --set roServers.image.tag=${DOCKER_VERSION} --set woServers.image.tag=${DOCKER_VERSION} --set expose.type=clusterIP -f ci/values.yaml --name ${env.JOB_NAME}-${env.BUILD_NUMBER}-cluster --namespace milvus-cluster --version 0.1.0 . "
+            sh "helm install --set roServers.image.tag=${DOCKER_VERSION} --set woServers.image.tag=${DOCKER_VERSION} --set expose.type=clusterIP -f ci/values.yaml --name ${env.JOB_NAME}-${env.BUILD_NUMBER}-cluster --namespace milvus-cluster --version 0.3.1 . "
         }
-        waitUntil {
-            def result = sh script: "nc -z -w 2 ${env.JOB_NAME}-${env.BUILD_NUMBER}-cluster-milvus-cluster-proxy.milvus-cluster.svc.cluster.local 19530", returnStatus: true
-            return !result
+        timeout(time: 2, unit: 'MINUTES') {
+            waitUntil {
+                def result = sh script: "nc -z -w 2 ${env.JOB_NAME}-${env.BUILD_NUMBER}-cluster-milvus-cluster-proxy.milvus-cluster.svc.cluster.local 19530", returnStatus: true
+                return !result
+            }
         }
     }
 } catch (exc) {

ci/jenkinsfile/deploy2dev.groovy

@@ -2,10 +2,17 @@ try {
     sh 'helm init --client-only --skip-refresh --stable-repo-url https://kubernetes.oss-cn-hangzhou.aliyuncs.com/charts'
     sh 'helm repo add milvus https://registry.zilliz.com/chartrepo/milvus'
     sh 'helm repo update'
-    sh "helm install --set engine.image.tag=${DOCKER_VERSION} --set expose.type=clusterIP --name ${env.JOB_NAME}-${env.BUILD_NUMBER} --version 0.3.0 milvus/milvus-gpu"
-    waitUntil {
-        def result = sh script: "nc -z -w 2 ${env.JOB_NAME}-${env.BUILD_NUMBER}-milvus-gpu-engine.kube-opt.svc.cluster.local 19530", returnStatus: true
-        return !result
+    dir ("milvus-helm") {
+        checkout([$class: 'GitSCM', branches: [[name: "${SEMVER}"]], doGenerateSubmoduleConfigurations: false, extensions: [], submoduleCfg: [], userRemoteConfigs: [[credentialsId: "${params.GIT_USER}", url: "git@192.168.1.105:megasearch/milvus-helm.git", name: 'origin', refspec: "+refs/heads/${SEMVER}:refs/remotes/origin/${SEMVER}"]]])
+        dir ("milvus/milvus-gpu") {
+            sh "helm install --set engine.image.tag=${DOCKER_VERSION} --set expose.type=clusterIP --name ${env.JOB_NAME}-${env.BUILD_NUMBER} --version 0.3.1 ."
+        }
+    }
+    timeout(time: 2, unit: 'MINUTES') {
+        waitUntil {
+            def result = sh script: "nc -z -w 2 ${env.JOB_NAME}-${env.BUILD_NUMBER}-milvus-gpu-engine.kube-opt.svc.cluster.local 19530", returnStatus: true
+            return !result
+        }
     }
 } catch (exc) {
     updateGitlabCommitStatus name: 'Deloy to Dev', state: 'failed'

ci/main_jenkinsfile

@@ -337,16 +337,18 @@ spec:
     post {
         always {
             script {
-                if (!currentBuild.resultIsBetterOrEqualTo('SUCCESS')) {
-                    // Send an email only if the build status has changed from green/unstable to red
-                    emailext subject: '$DEFAULT_SUBJECT',
-                    body: '$DEFAULT_CONTENT',
-                    recipientProviders: [
-                        [$class: 'DevelopersRecipientProvider'],
-                        [$class: 'RequesterRecipientProvider']
-                    ],
-                    replyTo: '$DEFAULT_REPLYTO',
-                    to: '$DEFAULT_RECIPIENTS'
+                if (env.gitlabAfter != null) {
+                    if (!currentBuild.resultIsBetterOrEqualTo('SUCCESS')) {
+                        // Send an email only if the build status has changed from green/unstable to red
+                        emailext subject: '$DEFAULT_SUBJECT',
+                        body: '$DEFAULT_CONTENT',
+                        recipientProviders: [
+                            [$class: 'DevelopersRecipientProvider'],
+                            [$class: 'RequesterRecipientProvider']
+                        ],
+                        replyTo: '$DEFAULT_REPLYTO',
+                        to: '$DEFAULT_RECIPIENTS'
+                    }
                 }
             }
         }

cpp/CHANGELOG.md

@@ -12,6 +12,7 @@ Please mark all change in change log and use the ticket from JIRA.
 - MS-153 - fix c_str error when connecting to MySQL
 - MS-157 - fix changelog
 - MS-190 - use env variable to switch mem manager and fix cmake
+- MS-217 - Fix SQ8 row count bug
 
 ## Improvement
 - MS-156 - Add unittest for merge result functions

cpp/conf/server_config.template

@@ -14,8 +14,8 @@ db_config:
   db_backend_url: sqlite://:@:/
 
   index_building_threshold: 1024   # index building trigger threshold, default: 1024, unit: MB
-  archive_disk_threshold: 512      # triger archive action if storage size exceed this value, unit: GB
-  archive_days_threshold: 30       # files older than x days will be archived, unit: day
+  archive_disk_threshold: 0        # triger archive action if storage size exceed this value, 0 means no limit, unit: GB
+  archive_days_threshold: 0        # files older than x days will be archived, 0 means no limit, unit: day
   maximum_memory: 4                # maximum memory allowed, default: 4, unit: GB, should be at least 1 GB.
                                    # the sum of maximum_memory and cpu_cache_capacity should be less than total memory
 
@@ -39,4 +39,4 @@ engine_config:
   nprobe: 10
   nlist: 16384
   use_blas_threshold: 20
-  metric_type: L2       #L2 or Inner Product
+  metric_type: L2       # compare vectors by euclidean distance(L2) or inner product(IP), optional: L2 or IP

cpp/src/cache/DataObj.h

@@ -20,6 +20,11 @@ public:
             : index_(index)
     {}
 
+    DataObj(const engine::Index_ptr& index, int64_t size)
+            : index_(index),
+              size_(size)
+    {}
+
     engine::Index_ptr data() { return index_; }
     const engine::Index_ptr& data() const { return index_; }
 
@@ -28,11 +33,16 @@ public:
             return 0;
         }
 
+        if(size_ > 0) {
+            return size_;
+        }
+
         return index_->ntotal*(index_->dim*4);
     }
 
 private:
     engine::Index_ptr index_ = nullptr;
+    int64_t size_ = 0;
 };
 
 using DataObjPtr = std::shared_ptr<DataObj>;

cpp/src/db/DBImpl.cpp

@@ -169,7 +169,10 @@ Status DBImpl::Query(const std::string& table_id, uint64_t k, uint64_t nq,
         }
     }
 
-    return QueryAsync(table_id, file_id_array, k, nq, vectors, dates, results);
+    cache::CpuCacheMgr::GetInstance()->PrintInfo(); //print cache info before query
+    status = QueryAsync(table_id, file_id_array, k, nq, vectors, dates, results);
+    cache::CpuCacheMgr::GetInstance()->PrintInfo(); //print cache info after query
+    return status;
 }
 
 Status DBImpl::Query(const std::string& table_id, const std::vector<std::string>& file_ids,
@@ -194,7 +197,10 @@ Status DBImpl::Query(const std::string& table_id, const std::vector<std::string>
         return Status::Error("Invalid file id");
     }
 
-    return QueryAsync(table_id, files_array, k, nq, vectors, dates, results);
+    cache::CpuCacheMgr::GetInstance()->PrintInfo(); //print cache info before query
+    status = QueryAsync(table_id, files_array, k, nq, vectors, dates, results);
+    cache::CpuCacheMgr::GetInstance()->PrintInfo(); //print cache info after query
+    return status;
 }
 
 Status DBImpl::QueryAsync(const std::string& table_id, const meta::TableFilesSchema& files,
@@ -486,7 +492,7 @@ Status DBImpl::BuildIndex(const meta::TableFileSchema& file) {
 
         //step 6: update meta
         table_file.file_type_ = meta::TableFileSchema::INDEX;
-        table_file.size_ = index->PhysicalSize();
+        table_file.size_ = index->Size();
 
         auto to_remove = file;
         to_remove.file_type_ = meta::TableFileSchema::TO_DELETE;

cpp/src/db/DBMetaImpl.cpp

@@ -674,16 +674,22 @@ Status DBMetaImpl::Archive() {
 Status DBMetaImpl::Size(uint64_t &result) {
     result = 0;
     try {
-        auto selected = ConnectorPtr->select(columns(sum(&TableFileSchema::size_)),
-                                             where(
-                                                 c(&TableFileSchema::file_type_) != (int) TableFileSchema::TO_DELETE
-                                             ));
+        auto files = ConnectorPtr->select(columns(&TableFileSchema::size_,
+                                                  &TableFileSchema::file_type_,
+                                                  &TableFileSchema::engine_type_),
+                                          where(
+                                                  c(&TableFileSchema::file_type_) != (int) TableFileSchema::TO_DELETE
+                                          ));
 
-        for (auto &sub_query : selected) {
-            if (!std::get<0>(sub_query)) {
-                continue;
+        for (auto &file : files) {
+            auto file_size = std::get<0>(file);
+            auto file_type = std::get<1>(file);
+            auto engine_type = std::get<2>(file);
+            if(file_type == (int)TableFileSchema::INDEX && engine_type == (int)EngineType::FAISS_IVFSQ8) {
+                result += (uint64_t)file_size/4;//hardcode for sq8
+            } else {
+                result += (uint64_t)file_size;
             }
-            result += (uint64_t) (*std::get<0>(sub_query));
         }
     } catch (std::exception &e) {
         return HandleException("Encounter exception when calculte db size", e);

cpp/src/db/FaissExecutionEngine.cpp

@@ -110,7 +110,7 @@ Status FaissExecutionEngine::Merge(const std::string& location) {
     if (location == location_) {
         return Status::Error("Cannot Merge Self");
     }
-    ENGINE_LOG_DEBUG << "Merge index file: " << location << " to: " << location_;
+    ENGINE_LOG_DEBUG << "Merge raw file: " << location << " to: " << location_;
 
     auto to_merge = zilliz::milvus::cache::CpuCacheMgr::GetInstance()->GetIndex(location);
     if (!to_merge) {
@@ -165,8 +165,9 @@ Status FaissExecutionEngine::Search(long n,
 }
 
 Status FaissExecutionEngine::Cache() {
-    zilliz::milvus::cache::CpuCacheMgr::GetInstance(
-            )->InsertItem(location_, std::make_shared<Index>(pIndex_));
+    auto index = std::make_shared<Index>(pIndex_);
+    cache::DataObjPtr data_obj = std::make_shared<cache::DataObj>(index, PhysicalSize());
+    zilliz::milvus::cache::CpuCacheMgr::GetInstance()->InsertItem(location_, data_obj);
 
     return Status::OK();
 }

cpp/src/db/Options.cpp

@@ -41,6 +41,10 @@ void ArchiveConf::ParseCritirias(const std::string& criterias) {
     }
 
     for (auto& token : tokens) {
+        if(token.empty()) {
+            continue;
+        }
+
         std::vector<std::string> kv;
         boost::algorithm::split(kv, token, boost::is_any_of(":"));
         if (kv.size() != 2) {

cpp/src/db/Options.h

@@ -22,7 +22,7 @@ static constexpr uint64_t ONE_GB = ONE_KB*ONE_MB;
 
 static const std::string ARCHIVE_CONF_DISK = "disk";
 static const std::string ARCHIVE_CONF_DAYS = "days";
-static const std::string ARCHIVE_CONF_DEFAULT = ARCHIVE_CONF_DISK + ":512";
+static const std::string ARCHIVE_CONF_DEFAULT = "";
 
 struct ArchiveConf {
     using CriteriaT = std::map<std::string, int>;

cpp/src/db/scheduler/task/SearchTask.cpp

@@ -107,7 +107,7 @@ Status SearchTask::ClusterResult(const std::vector<long> &output_ids,
                                  uint64_t nq,
                                  uint64_t topk,
                                  SearchContext::ResultSet &result_set) {
-    if(output_ids.size() != nq*topk || output_distence.size() != nq*topk) {
+    if(output_ids.size() < nq*topk || output_distence.size() < nq*topk) {
         std::string msg = "Invalid id array size: " + std::to_string(output_ids.size()) +
                 " distance array size: " + std::to_string(output_distence.size());
         SERVER_LOG_ERROR << msg;

cpp/src/sdk/examples/simple/src/ClientTest.cpp

@@ -23,6 +23,7 @@ namespace {
     static constexpr int64_t TOP_K = 10;
     static constexpr int64_t SEARCH_TARGET = 5000; //change this value, result is different
     static constexpr int64_t ADD_VECTOR_LOOP = 10;
+    static constexpr int64_t SECONDS_EACH_HOUR = 3600;
 
 #define BLOCK_SPLITER std::cout << "===========================================" << std::endl;
 
@@ -59,7 +60,7 @@ namespace {
     std::string CurrentTime() {
         time_t tt;
         time( &tt );
-        tt = tt + 8*3600;
+        tt = tt + 8*SECONDS_EACH_HOUR;
         tm* t= gmtime( &tt );
 
         std::string str = std::to_string(t->tm_year + 1900) + "_" + std::to_string(t->tm_mon + 1)
@@ -69,10 +70,11 @@ namespace {
         return str;
     }
 
-    std::string CurrentTmDate() {
+    std::string CurrentTmDate(int64_t offset_day = 0) {
         time_t tt;
         time( &tt );
-        tt = tt + 8*3600;
+        tt = tt + 8*SECONDS_EACH_HOUR;
+        tt = tt + 24*SECONDS_EACH_HOUR*offset_day;
         tm* t= gmtime( &tt );
 
         std::string str = std::to_string(t->tm_year + 1900) + "-" + std::to_string(t->tm_mon + 1)
@@ -160,7 +162,7 @@ namespace {
         std::vector<Range> query_range_array;
         Range rg;
         rg.start_value = CurrentTmDate();
-        rg.end_value = CurrentTmDate();
+        rg.end_value = CurrentTmDate(1);
         query_range_array.emplace_back(rg);
 
         std::vector<RowRecord> record_array;

cpp/src/wrapper/Operand.cpp

@@ -28,6 +28,15 @@ IndexType resolveIndexType(const string &index_type) {
     return IndexType::Invalid_Option;
 }
 
+int CalcBacketCount(int nb, size_t nlist) {
+    int backet_count = int(nb / 1000000.0 * nlist);
+    if(backet_count == 0) {
+        backet_count = 1; //avoid faiss rash
+    }
+
+    return backet_count;
+}
+
 // nb at least 100
 string Operand::get_index_type(const int &nb) {
     if (!index_str.empty()) { return index_str; }
@@ -45,7 +54,7 @@ string Operand::get_index_type(const int &nb) {
             size_t nlist = engine_config.GetInt32Value(CONFIG_NLIST, 16384);
 
             index_str += (ncent != 0 ? index_type + std::to_string(ncent) :
-                          index_type + std::to_string(int(nb / 1000000.0 * nlist)));
+                          index_type + std::to_string(CalcBacketCount(nb, nlist)));
 //            std::cout<<"nlist = "<<nlist<<std::endl;
             if (!postproc.empty()) { index_str += ("," + postproc); }
             break;
@@ -58,7 +67,7 @@ string Operand::get_index_type(const int &nb) {
             size_t nlist = engine_config.GetInt32Value(CONFIG_NLIST, 16384);
 
             index_str += (ncent != 0 ? "IVF" + std::to_string(ncent) :
-                          "IVF" + std::to_string(int(nb / 1000000.0 * nlist)));
+                          "IVF" + std::to_string(CalcBacketCount(nb, nlist)));
             index_str += ",SQ8";
 //            std::cout<<"nlist = "<<nlist<<std::endl;
             break;