Update build environment

Signed-off-by: quicksilver <zhifeng.zhang@zilliz.com>
pull/4973/head^2
quicksilver 2020-10-26 15:45:18 +08:00 committed by yefu.chen
parent a48ca80286
commit eb64839aef
32 changed files with 501 additions and 523 deletions

3
.env Normal file
View File

@ -0,0 +1,3 @@
REPO=milvusdb/milvus-distributed-dev
ARCH=amd64
UBUNTU=18.04

View File

@ -26,41 +26,46 @@ on:
jobs:
ubuntu:
name: AMD64 ubuntu-18.04
runs-on: ubuntu-18.04
name: AMD64 Ubuntu ${{ matrix.ubuntu }}
runs-on: ubuntu-latest
timeout-minutes: 60
strategy:
fail-fast: false
matrix:
ubuntu: [18.04]
env:
UBUNTU: ${{ matrix.ubuntu }}
steps:
- name: Checkout
uses: actions/checkout@v2
- name: Install Dependency
run: |
./scripts/install_deps.sh
go get github.com/golang/protobuf/protoc-gen-go@v1.3.2
- name: Cache Core Thirdparty
id: cache-core
uses: actions/cache@v2
- name: Check Dockerfile
uses: reviewdog/action-hadolint@v1
with:
path: |
./internal/core/cmake_build
key: ${{ runner.os }}-core-thirdparty
- name: Build Cpp
github_token: ${{ secrets.GITHUB_TOKEN }}
reporter: github-pr-check # Default is github-pr-check
- name: Docker Pull
shell: bash
run: |
./scripts/core_build.sh -u
- name: Generat Proto GO File
docker-compose pull --ignore-pull-failures ubuntu
- name: Docker Build
shell: bash
run: |
echo `pwd`
pwd_dir=`pwd`
export PATH=$PATH:$(go env GOPATH)/bin
export protoc=${pwd_dir}/internal/core/cmake_build/thirdparty/protobuf/protobuf-build/protoc
./scripts/proto_gen_go.sh
- name: Build GO
docker-compose build ubuntu
docker rmi $(docker images | grep '<none>' | awk '{print $3}') || exit 0
- name: Cache Docker Volumes
uses: actions/cache@v1
with:
path: .docker
key: ubuntu${{ matrix.ubuntu }}-${{ hashFiles('internal/core/**') }}
restore-keys: ubuntu${{ matrix.ubuntu }}-
- name: Docker Run
run: |
go build -o ./cmd/writer/writer ./cmd/writer/writer.go
go build -o ./cmd/reader/reader ./cmd/reader/reader.go
go build -o ./cmd/master/master ./cmd/master/master.go
go build -o ./cmd/proxy/proxy ./cmd/proxy/proxy.go
- name: Docker Pull And Run
docker-compose run ubuntu
- name: Docker Push
if: success() && github.event_name == 'push' && github.repository == 'zilliztech/milvus-distributed'
continue-on-error: true
shell: bash
run: |
docker-compose up -d
- name: Run Unittest
run: |
./scripts/run_unittest.sh
docker login -u ${{ secrets.DOCKERHUB_USER }} \
-p ${{ secrets.DOCKERHUB_TOKEN }}
docker-compose push ubuntu

View File

@ -42,7 +42,7 @@ RUN mkdir -p /usr/local/go && wget -qO- "https://golang.org/dl/go1.15.2.linux-am
go get github.com/golang/protobuf/protoc-gen-go@v1.3.2
# Set permissions on /etc/passwd and /home to allow arbitrary users to write
COPY --chown=0:0 docker/build_env/entrypoint.sh /
COPY --chown=0:0 build/docker/env/entrypoint.sh /
RUN mkdir -p /home/user && chgrp -R 0 /home && chmod -R g=u /etc/passwd /etc/group /home && chmod +x /entrypoint.sh
ENV HOME=/home/user

View File

@ -25,12 +25,12 @@ func init() {
// go mock.FakePulsarProducer()
}
func main() {
var yamlFile string
flag.StringVar(&yamlFile, "yaml", "", "yaml file")
flag.Parse()
// flag.Usage()
fmt.Println("yaml file: ", yamlFile)
conf.LoadConfig(yamlFile)
var yamlFile string
flag.StringVar(&yamlFile, "yaml", "", "yaml file")
flag.Parse()
// flag.Usage()
fmt.Println("yaml file: ", yamlFile)
conf.LoadConfig(yamlFile)
master.Run()
//master.SegmentStatsController()

View File

@ -1,60 +1,25 @@
version: '3.5'
x-ccache: &ccache
CCACHE_COMPILERCHECK: content
CCACHE_COMPRESS: 1
CCACHE_COMPRESSLEVEL: 5
CCACHE_MAXSIZE: 2G
CCACHE_DIR: /ccache
services:
etcd:
image: quay.io/coreos/etcd:latest
command: etcd -listen-peer-urls=http://127.0.0.1:12380 -advertise-client-urls=http://127.0.0.1:12379 -listen-client-urls http://0.0.0.0:12379,http://0.0.0.0:14001 -initial-advertise-peer-urls=http://127.0.0.1:12380 --initial-cluster default=http://127.0.0.1:12380
ports:
- "12379:12379"
- "12380:12380"
- "14001:14001"
pulsar:
image: apachepulsar/pulsar:latest
command: bin/pulsar standalone
ports:
- "6650:6650"
- "18080:8080"
pd0:
image: pingcap/pd:latest
network_mode: "host"
ports:
- "2379:2379"
- "2380:2380"
volumes:
- /tmp/config/pd.toml:/pd.toml:ro
- /tmp/data:/data
- /tmp/logs:/logs
- /etc/localtime:/etc/localtime:ro
command:
- --name=pd0
- --client-urls=http://0.0.0.0:2379
- --peer-urls=http://0.0.0.0:2380
- --advertise-client-urls=http://127.0.0.1:2379
- --advertise-peer-urls=http://127.0.0.1:2380
- --initial-cluster=pd0=http://127.0.0.1:2380
- --data-dir=/data/pd0
- --log-file=/logs/pd0.log
restart: on-failure
tikv0:
network_mode: "host"
image: pingcap/tikv:latest
ports:
- "20160:20160"
volumes:
- /tmp/config/tikv.toml:/tikv.toml:ro
- /tmp/data:/data
- /tmp/logs:/logs
- /etc/localtime:/etc/localtime:ro
command:
- --addr=0.0.0.0:20160
- --advertise-addr=127.0.0.1:20160
- --data-dir=/data/tikv0
- --pd=127.0.0.1:2379
- --log-file=/logs/tikv0.log
depends_on:
- "pd0"
restart: on-failure
ubuntu:
image: ${REPO}:${ARCH}-ubuntu${UBUNTU}
build:
context: .
dockerfile: build/docker/env/cpu/ubuntu${UBUNTU}/Dockerfile
cache_from:
- ${REPO}:${ARCH}-ubuntu${UBUNTU}
shm_size: 2G
environment:
<<: *ccache
volumes: &ubuntu-volumes
- .:/milvus-distributed:delegated
- ${DOCKER_VOLUME_DIRECTORY:-.docker}/${ARCH}-ubuntu${UBUNTU}-cache:/ccache:delegated
command: &ubuntu-command
["/bin/bash", "-l", "-c", "/milvus-distributed/scripts/core_build.sh -u"]

View File

@ -5,10 +5,19 @@ if [[ ! ${jobs+1} ]]; then
jobs=$(nproc)
fi
SOURCE="${BASH_SOURCE[0]}"
while [ -h "$SOURCE" ]; do # resolve $SOURCE until the file is no longer a symlink
DIR="$( cd -P "$( dirname "$SOURCE" )" && pwd )"
SOURCE="$(readlink "$SOURCE")"
[[ $SOURCE != /* ]] && SOURCE="$DIR/$SOURCE" # if $SOURCE was a relative symlink, we need to resolve it relative to the path where the symlink file was located
done
SCRIPTS_DIR="$( cd -P "$( dirname "$SOURCE" )" && pwd )"
BUILD_OUTPUT_DIR="cmake_build"
BUILD_TYPE="Release"
BUILD_UNITTEST="OFF"
INSTALL_PREFIX=$(pwd)/output
INSTALL_PREFIX=${SCRIPTS_DIR}/output
MAKE_CLEAN="OFF"
BUILD_COVERAGE="OFF"
DB_PATH="/tmp/milvus"

View File

@ -1,58 +0,0 @@
find_package(Threads REQUIRED)
include(ExternalProject)
ExternalProject_Add(
googletest
URL http://ss2.fluorinedog.com/data/gtest_v1.10.x.zip
UPDATE_COMMAND ""
INSTALL_COMMAND ""
LOG_DOWNLOAD ON
LOG_CONFIGURE ON
LOG_BUILD ON)
ExternalProject_Get_Property(googletest source_dir)
set(GTEST_INCLUDE_DIRS ${source_dir}/googletest/include)
set(GMOCK_INCLUDE_DIRS ${source_dir}/googlemock/include)
# The cloning of the above repo doesn't happen until make, however if the dir doesn't
# exist, INTERFACE_INCLUDE_DIRECTORIES will throw an error.
# To make it work, we just create the directory now during config.
file(MAKE_DIRECTORY ${GTEST_INCLUDE_DIRS})
file(MAKE_DIRECTORY ${GMOCK_INCLUDE_DIRS})
ExternalProject_Get_Property(googletest binary_dir)
set(GTEST_LIBRARY_PATH ${binary_dir}/lib/${CMAKE_FIND_LIBRARY_PREFIXES}gtest.a)
set(GTEST_LIBRARY gtest)
add_library(${GTEST_LIBRARY} UNKNOWN IMPORTED)
set_target_properties(${GTEST_LIBRARY} PROPERTIES
"IMPORTED_LOCATION" "${GTEST_LIBRARY_PATH}"
"IMPORTED_LINK_INTERFACE_LIBRARIES" "${CMAKE_THREAD_LIBS_INIT}"
"INTERFACE_INCLUDE_DIRECTORIES" "${GTEST_INCLUDE_DIRS}")
add_dependencies(${GTEST_LIBRARY} googletest)
set(GTEST_MAIN_LIBRARY_PATH ${binary_dir}/lib/${CMAKE_FIND_LIBRARY_PREFIXES}gtest_main.a)
set(GTEST_MAIN_LIBRARY gtest_main)
add_library(${GTEST_MAIN_LIBRARY} UNKNOWN IMPORTED)
set_target_properties(${GTEST_MAIN_LIBRARY} PROPERTIES
"IMPORTED_LOCATION" "${GTEST_MAIN_LIBRARY_PATH}"
"IMPORTED_LINK_INTERFACE_LIBRARImS" "${CMAKE_THREAD_LIBS_INIT}"
"INTERFACE_INCLUDE_DIRECTORIES" "${GTEST_INCLUDE_DIRS}")
add_dependencies(${GTEST_MAIN_LIBRARY} googletest)
# set(GMOCK_LIBRARY_PATH ${binary_dir}/googlemock/${CMAKE_FIND_LIBRARY_PREFIXES}gmock.a)
# set(GMOCK_LIBRARY gmock)
# add_library(${GMOCK_LIBRARY} UNKNOWN IMPORTED)
# set_target_properties(${GMOCK_LIBRARY} PROPERTIES
# "IMPORTED_LOCATION" "${GMOCK_LIBRARY_PATH}"
# "IMPORTED_LINK_INTERFACE_LIBRARIES" "${CMAKE_THREAD_LIBS_INIT}"
# "INTERFACE_INCLUDE_DIRECTORIES" "${GMOCK_INCLUDE_DIRS}")
# add_dependencies(${GMOCK_LIBRARY} googletest)
# set(GMOCK_MAIN_LIBRARY_PATH ${binary_dir}/googlemock/${CMAKE_FIND_LIBRARY_PREFIXES}gmock_main.a)
# set(GMOCK_MAIN_LIBRARY gmock_main)
# add_library(${GMOCK_MAIN_LIBRARY} UNKNOWN IMPORTED)
# set_target_properties(${GMOCK_MAIN_LIBRARY} PROPERTIES
# "IMPORTED_LOCATION" "${GMOCK_MAIN_LIBRARY_PATH}"
# "IMPORTED_LINK_INTERFACE_LIBRARIES" "${CMAKE_THREAD_LIBS_INIT}"
# "INTERFACE_INCLUDE_DIRECTORIES" "${GMOCK_INCLUDE_DIRS}")
# add_dependencies(${GMOCK_MAIN_LIBRARY} ${GTEST_LIBRARY})

View File

@ -29,47 +29,3 @@ add_subdirectory( log)
add_subdirectory( dog_segment)
add_subdirectory( cache )
add_subdirectory( query )
# add_subdirectory( db ) # target milvus_engine
# add_subdirectory( server )
# set(link_lib
# milvus_engine
# # dog_segment
# #query
# utils
# curl
# )
# set( BOOST_LIB libboost_system.a
# libboost_filesystem.a
# libboost_serialization.a
# )
# set( THIRD_PARTY_LIBS yaml-cpp
# )
# target_link_libraries( server
# PUBLIC ${link_lib}
# ${THIRD_PARTY_LIBS}
# ${BOOST_LIB}
# )
# # **************************** Get&Print Include Directories ****************************
# get_property( dirs DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR} PROPERTY INCLUDE_DIRECTORIES )
# foreach ( dir ${dirs} )
# message( STATUS "Current Include DIRS: ")
# endforeach ()
# set( SERVER_LIBS server )
# add_executable( milvus_server ${CMAKE_CURRENT_SOURCE_DIR}/main.cpp
# )
# #target_include_directories(db PUBLIC ${PROJECT_BINARY_DIR}/thirdparty/pulsar-client-cpp/pulsar-client-cpp-src/pulsar-client-cpp/include)
# target_link_libraries( milvus_server PRIVATE ${SERVER_LIBS} )
# install( TARGETS milvus_server DESTINATION bin )

View File

@ -14,7 +14,6 @@
set(KNOWHERE_THIRDPARTY_DEPENDENCIES
Arrow
FAISS
GTest
OpenBLAS
MKL
)
@ -31,8 +30,6 @@ endforeach ()
macro(build_dependency DEPENDENCY_NAME)
if ("${DEPENDENCY_NAME}" STREQUAL "Arrow")
build_arrow()
elseif ("${DEPENDENCY_NAME}" STREQUAL "GTest")
find_package(GTest REQUIRED)
elseif ("${DEPENDENCY_NAME}" STREQUAL "OpenBLAS")
build_openblas()
elseif ("${DEPENDENCY_NAME}" STREQUAL "FAISS")
@ -215,12 +212,6 @@ else ()
)
endif ()
# if (DEFINED ENV{KNOWHERE_GTEST_URL})
# set(GTEST_SOURCE_URL "$ENV{KNOWHERE_GTEST_URL}")
# else ()
# set(GTEST_SOURCE_URL
# "https://github.com/google/googletest/archive/release-${GTEST_VERSION}.tar.gz")
# endif ()
if (DEFINED ENV{KNOWHERE_OPENBLAS_URL})
set(OPENBLAS_SOURCE_URL "$ENV{KNOWHERE_OPENBLAS_URL}")
@ -383,91 +374,6 @@ if (KNOWHERE_WITH_OPENBLAS)
link_directories(SYSTEM "${OpenBLAS_LIB_DIR}")
endif()
# ----------------------------------------------------------------------
# Google gtest
# macro(build_gtest)
# message(STATUS "Building gtest-${GTEST_VERSION} from source")
# set(GTEST_VENDORED TRUE)
# set(GTEST_CMAKE_CXX_FLAGS "${EP_CXX_FLAGS}")
#
# if (APPLE)
# set(GTEST_CMAKE_CXX_FLAGS
# ${GTEST_CMAKE_CXX_FLAGS}
# -DGTEST_USE_OWN_TR1_TUPLE=1
# -Wno-unused-value
# -Wno-ignored-attributes)
# endif ()
#
# set(GTEST_PREFIX "${INDEX_BINARY_DIR}/googletest_ep-prefix/src/googletest_ep")
# set(GTEST_INCLUDE_DIR "${GTEST_PREFIX}/include")
# set(GTEST_STATIC_LIB
# "${GTEST_PREFIX}/lib/${CMAKE_STATIC_LIBRARY_PREFIX}gtest${CMAKE_STATIC_LIBRARY_SUFFIX}")
# set(GTEST_MAIN_STATIC_LIB
# "${GTEST_PREFIX}/lib/${CMAKE_STATIC_LIBRARY_PREFIX}gtest_main${CMAKE_STATIC_LIBRARY_SUFFIX}")
#
# set(GTEST_CMAKE_ARGS
# ${EP_COMMON_CMAKE_ARGS}
# "-DCMAKE_INSTALL_PREFIX=${GTEST_PREFIX}"
# "-DCMAKE_INSTALL_LIBDIR=lib"
# -DCMAKE_CXX_FLAGS=${GTEST_CMAKE_CXX_FLAGS}
# -DCMAKE_BUILD_TYPE=Release)
#
# set(GMOCK_INCLUDE_DIR "${GTEST_PREFIX}/include")
# set(GMOCK_STATIC_LIB
# "${GTEST_PREFIX}/lib/${CMAKE_STATIC_LIBRARY_PREFIX}gmock${CMAKE_STATIC_LIBRARY_SUFFIX}"
# )
#
# ExternalProject_Add(googletest_ep
# URL
# ${GTEST_SOURCE_URL}
# BUILD_COMMAND
# ${MAKE}
# ${MAKE_BUILD_ARGS}
# BUILD_BYPRODUCTS
# ${GTEST_STATIC_LIB}
# ${GTEST_MAIN_STATIC_LIB}
# ${GMOCK_STATIC_LIB}
# CMAKE_ARGS
# ${GTEST_CMAKE_ARGS}
# ${EP_LOG_OPTIONS})
#
# # The include directory must exist before it is referenced by a target.
# file(MAKE_DIRECTORY "${GTEST_INCLUDE_DIR}")
#
# add_library(gtest STATIC IMPORTED)
# set_target_properties(gtest
# PROPERTIES IMPORTED_LOCATION "${GTEST_STATIC_LIB}"
# INTERFACE_INCLUDE_DIRECTORIES "${GTEST_INCLUDE_DIR}")
#
# add_library(gtest_main STATIC IMPORTED)
# set_target_properties(gtest_main
# PROPERTIES IMPORTED_LOCATION "${GTEST_MAIN_STATIC_LIB}"
# INTERFACE_INCLUDE_DIRECTORIES "${GTEST_INCLUDE_DIR}")
#
# add_library(gmock STATIC IMPORTED)
# set_target_properties(gmock
# PROPERTIES IMPORTED_LOCATION "${GMOCK_STATIC_LIB}"
# INTERFACE_INCLUDE_DIRECTORIES "${GTEST_INCLUDE_DIR}")
#
# add_dependencies(gtest googletest_ep)
# add_dependencies(gtest_main googletest_ep)
# add_dependencies(gmock googletest_ep)
#
# endmacro()
## if (KNOWHERE_BUILD_TESTS AND NOT TARGET googletest_ep)
#if ( NOT TARGET gtest AND KNOWHERE_BUILD_TESTS )
# resolve_dependency(GTest)
#
# if (NOT GTEST_VENDORED)
# endif ()
#
# # TODO: Don't use global includes but rather target_include_directories
# get_target_property(GTEST_INCLUDE_DIR gtest INTERFACE_INCLUDE_DIRECTORIES)
# link_directories(SYSTEM "${GTEST_PREFIX}/lib")
# include_directories(SYSTEM ${GTEST_INCLUDE_DIR})
#endif ()
# ----------------------------------------------------------------------
# MKL

View File

@ -1,50 +0,0 @@
// Copyright (C) 2019-2020 Zilliz. All rights reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software distributed under the License
// is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
// or implied. See the License for the specific language governing permissions and limitations under the License.
#include <getopt.h>
#include <unistd.h>
#include <csignal>
#include <cstring>
#include <string>
#include "config/ConfigMgr.h"
#include "easyloggingpp/easylogging++.h"
#include "utils/SignalHandler.h"
#include "utils/Status.h"
INITIALIZE_EASYLOGGINGPP
void
print_help(const std::string& app_name) {
std::cout << std::endl << "Usage: " << app_name << " [OPTIONS]" << std::endl;
std::cout << R"(
Options:
-h --help Print this help.
-c --conf_file filename Read configuration from the file.
-d --daemon Daemonize this application.
-p --pid_file filename PID file used by daemonized app.
)" << std::endl;
}
void
print_banner() {
std::cout << std::endl;
std::cout << " __ _________ _ ____ ______ " << std::endl;
std::cout << " / |/ / _/ /| | / / / / / __/ " << std::endl;
std::cout << " / /|_/ // // /_| |/ / /_/ /\\ \\ " << std::endl;
std::cout << " /_/ /_/___/____/___/\\____/___/ " << std::endl;
std::cout << std::endl;
}
int
main(int argc, char* argv[]) {
print_banner();
}

View File

@ -32,6 +32,12 @@ set( FETCHCONTENT_QUIET OFF )
set( THREADS_PREFER_PTHREAD_FLAG ON )
find_package( Threads REQUIRED )
# ****************************** Thirdparty googletest ***************************************
if ( MILVUS_BUILD_TESTS )
add_subdirectory( gtest )
endif()
# ****************************** Thirdparty yaml ***************************************
if ( MILVUS_WITH_YAMLCPP )
add_subdirectory( yaml-cpp )

View File

@ -0,0 +1,64 @@
#-------------------------------------------------------------------------------
# Copyright (C) 2019-2020 Zilliz. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software distributed under the License
# is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
# or implied. See the License for the specific language governing permissions and limitations under the License.
#-------------------------------------------------------------------------------
if ( DEFINED ENV{MILVUS_GTEST_URL} )
set( GTEST_SOURCE_URL "$ENV{MILVUS_GTEST_URL}" )
else()
set( GTEST_SOURCE_URL "https://gitee.com/quicksilver/googletest/repository/archive/release-${GTEST_VERSION}.zip" )
endif()
message( STATUS "Building gtest-${GTEST_VERSION} from source" )
set( CMAKE_POLICY_DEFAULT_CMP0022 NEW ) # for googletest only
FetchContent_Declare(
googletest
URL ${GTEST_SOURCE_URL}
URL_MD5 "f9137c5bc18b7d74027936f0f1bfa5c8"
SOURCE_DIR ${CMAKE_CURRENT_BINARY_DIR}/googletest-src
BINARY_DIR ${CMAKE_CURRENT_BINARY_DIR}/googletest-build
DOWNLOAD_DIR ${THIRDPARTY_DOWNLOAD_PATH} )
FetchContent_GetProperties( googletest )
if ( NOT googletest_POPULATED )
FetchContent_Populate( googletest )
# Adding the following targets:
# gtest, gtest_main, gmock, gmock_main
add_subdirectory( ${googletest_SOURCE_DIR}
${googletest_BINARY_DIR}
EXCLUDE_FROM_ALL )
endif()
# ****************************************************************
# Create ALIAS Target
# ****************************************************************
# if (NOT TARGET GTest:gtest)
# add_library( GTest::gtest ALIAS gtest )
# endif()
# if (NOT TARGET GTest:main)
# add_library( GTest::main ALIAS gtest_main )
# endif()
# if (NOT TARGET GMock:gmock)
# target_link_libraries( gmock INTERFACE GTest::gtest )
# add_library( GMock::gmock ALIAS gmock )
# endif()
# if (NOT TARGET GMock:main)
# target_link_libraries( gmock_main INTERFACE GTest::gtest )
# add_library( GMock::main ALIAS gmock_main )
# endif()
get_property( var DIRECTORY "${CMAKE_CURRENT_BINARY_DIR}/googletest-src" PROPERTY COMPILE_OPTIONS )
message( STATUS "gtest compile options: ${var}" )

View File

@ -31,6 +31,7 @@ FetchContent_Declare(
)
set( protobuf_BUILD_TESTS CACHE BOOL OFF FORCE )
if ( NOT protobuf_POPULATED )
FetchContent_Populate( protobuf )

View File

@ -1,8 +1,7 @@
enable_testing()
find_package(GTest REQUIRED)
include_directories(${CMAKE_HOME_DIRECTORY}/src)
include_directories(>>>> ${CMAKE_HOME_DIRECTORY}/src/index/knowhere)
include_directories(${CMAKE_HOME_DIRECTORY}/src/index/knowhere)
set(MILVUS_TEST_FILES
test_naive.cpp
test_dog_segment.cpp
@ -22,4 +21,4 @@ target_link_libraries(all_tests
log
pthread
)
install (TARGETS all_tests DESTINATION unittest)
install (TARGETS all_tests DESTINATION unittest)

View File

@ -5,7 +5,6 @@ import (
"github.com/zilliztech/milvus-distributed/internal/proto/etcdpb"
"github.com/zilliztech/milvus-distributed/internal/proto/schemapb"
messagepb "github.com/zilliztech/milvus-distributed/internal/proto/message"
"github.com/gogo/protobuf/proto"
jsoniter "github.com/json-iterator/go"
)
@ -21,12 +20,11 @@ type Collection struct {
SegmentIDs []uint64 `json:"segment_ids"`
PartitionTags []string `json:"partition_tags"`
GrpcMarshalString string `json:"grpc_marshal_string"`
IndexParam []*messagepb.IndexParam `json:"index_param"`
}
type FieldMeta struct {
FieldName string `json:"field_name"`
Type messagepb.DataType `json:"type"`
Type schemapb.DataType `json:"type"`
DIM int64 `json:"dimension"`
}
@ -59,12 +57,12 @@ func GrpcMarshal(c *Collection) *Collection {
}
func NewCollection(id uint64, name string, createTime time.Time,
schema []*messagepb.FieldMeta, sIds []uint64, ptags []string) Collection {
schema []*schemapb.FieldSchema, sIds []uint64, ptags []string) Collection {
segementIDs := []uint64{}
newSchema := []FieldMeta{}
for _, v := range schema {
newSchema = append(newSchema, FieldMeta{FieldName: v.FieldName, Type: v.Type, DIM: v.Dim})
newSchema = append(newSchema, FieldMeta{FieldName: v.Name, Type: v.DataType, DIM: 16})
}
for _, sid := range sIds {
segementIDs = append(segementIDs, sid)

View File

@ -3,8 +3,7 @@ package collection
import (
"testing"
"time"
messagepb "github.com/zilliztech/milvus-distributed/internal/proto/message"
"github.com/zilliztech/milvus-distributed/internal/proto/schemapb"
"github.com/stretchr/testify/assert"
)
@ -12,7 +11,7 @@ var (
cid = uint64(10011111234)
name = "test-segment"
createTime = time.Now()
schema = []*messagepb.FieldMeta{}
schema = []*schemapb.FieldSchema{}
sIds = []uint64{111111, 222222}
ptags = []string{"default", "test"}
)
@ -23,9 +22,8 @@ func TestNewCollection(t *testing.T) {
assert.Equal(cid, c.ID)
assert.Equal(name, c.Name)
for k, v := range schema {
assert.Equal(v.Dim, c.Schema[k].DIM)
assert.Equal(v.FieldName, c.Schema[k].FieldName)
assert.Equal(v.Type, c.Schema[k].Type)
assert.Equal(v.Name, c.Schema[k].FieldName)
assert.Equal(v.DataType, c.Schema[k].Type)
}
assert.Equal(sIds, c.SegmentIDs)
assert.Equal(ptags, c.PartitionTags)

View File

@ -7,27 +7,27 @@ import (
"github.com/zilliztech/milvus-distributed/internal/conf"
"github.com/zilliztech/milvus-distributed/internal/master/collection"
messagepb "github.com/zilliztech/milvus-distributed/internal/proto/message"
"github.com/zilliztech/milvus-distributed/internal/proto/schemapb"
"github.com/zilliztech/milvus-distributed/internal/master/id"
"github.com/zilliztech/milvus-distributed/internal/master/kv"
"github.com/zilliztech/milvus-distributed/internal/master/segment"
)
func CollectionController(ch chan *messagepb.Mapping, kvbase kv.Base, errch chan error) {
func CollectionController(ch chan *schemapb.CollectionSchema, kvbase kv.Base, errch chan error) {
for collectionMeta := range ch {
sID := id.New().Uint64()
cID := id.New().Uint64()
s2ID := id.New().Uint64()
fieldMetas := []*messagepb.FieldMeta{}
if collectionMeta.Schema != nil {
fieldMetas = collectionMeta.Schema.FieldMetas
fieldMetas := []*schemapb.FieldSchema{}
if collectionMeta.Fields != nil {
fieldMetas = collectionMeta.Fields
}
c := collection.NewCollection(cID, collectionMeta.CollectionName,
c := collection.NewCollection(cID, collectionMeta.Name,
time.Now(), fieldMetas, []uint64{sID, s2ID},
[]string{"default"})
cm := collection.GrpcMarshal(&c)
s := segment.NewSegment(sID, cID, collectionMeta.CollectionName, "default", 0, 511, time.Now(), time.Unix(1<<36-1, 0))
s2 := segment.NewSegment(s2ID, cID, collectionMeta.CollectionName, "default", 512, 1023, time.Now(), time.Unix(1<<36-1, 0))
s := segment.NewSegment(sID, cID, collectionMeta.Name, "default", 0, 511, time.Now(), time.Unix(1<<36-1, 0))
s2 := segment.NewSegment(s2ID, cID, collectionMeta.Name, "default", 512, 1023, time.Now(), time.Unix(1<<36-1, 0))
collectionData, _ := collection.Collection2JSON(*cm)
segmentData, err := segment.Segment2JSON(s)
if err != nil {
@ -52,18 +52,18 @@ func CollectionController(ch chan *messagepb.Mapping, kvbase kv.Base, errch chan
}
}
func WriteCollection2Datastore(collectionMeta *messagepb.Mapping, kvbase kv.Base) error {
func WriteCollection2Datastore(collectionMeta *schemapb.CollectionSchema, kvbase kv.Base) error {
sID := id.New().Uint64()
cID := id.New().Uint64()
fieldMetas := []*messagepb.FieldMeta{}
if collectionMeta.Schema != nil {
fieldMetas = collectionMeta.Schema.FieldMetas
fieldMetas := []*schemapb.FieldSchema{}
if collectionMeta.Fields != nil {
fieldMetas = collectionMeta.Fields
}
c := collection.NewCollection(cID, collectionMeta.CollectionName,
c := collection.NewCollection(cID, collectionMeta.Name,
time.Now(), fieldMetas, []uint64{sID},
[]string{"default"})
cm := collection.GrpcMarshal(&c)
s := segment.NewSegment(sID, cID, collectionMeta.CollectionName, "default", 0, conf.Config.Pulsar.TopicNum, time.Now(), time.Unix(1<<46-1, 0))
s := segment.NewSegment(sID, cID, collectionMeta.Name, "default", 0, conf.Config.Pulsar.TopicNum, time.Now(), time.Unix(1<<46-1, 0))
collectionData, err := collection.Collection2JSON(*cm)
if err != nil {
log.Fatal(err)
@ -87,31 +87,3 @@ func WriteCollection2Datastore(collectionMeta *messagepb.Mapping, kvbase kv.Base
return nil
}
func UpdateCollectionIndex(index *messagepb.IndexParam, kvbase kv.Base) error {
collectionName := index.CollectionName
collectionJSON, err := kvbase.Load("collection/" + collectionName)
if err != nil {
return err
}
c, err := collection.JSON2Collection(collectionJSON)
if err != nil {
return err
}
for k, v := range c.IndexParam {
if v.IndexName == index.IndexName {
c.IndexParam[k] = v
}
}
c.IndexParam = append(c.IndexParam, index)
cm := collection.GrpcMarshal(c)
collectionData, err := collection.Collection2JSON(*cm)
if err != nil {
return err
}
err = kvbase.Save("collection/"+collectionName, collectionData)
if err != nil {
return err
}
return nil
}

View File

@ -7,7 +7,7 @@ import (
"github.com/zilliztech/milvus-distributed/internal/conf"
"github.com/zilliztech/milvus-distributed/internal/master/kv"
"github.com/zilliztech/milvus-distributed/internal/master/segment"
"github.com/zilliztech/milvus-distributed/internal/proto/internalpb"
"go.etcd.io/etcd/clientv3"
)
@ -25,12 +25,11 @@ func newKvBase() kv.Base {
func TestComputeClosetTime(t *testing.T) {
kvbase := newKvBase()
var news segment.SegmentStats
var news internalpb.SegmentStatistics
for i := 0; i < 10; i++ {
news = segment.SegmentStats{
SegementID: uint64(6875940398055133887),
news = internalpb.SegmentStatistics{
SegmentId: uint64(6875940398055133887),
MemorySize: uint64(i * 1000),
MemoryRate: 0.9,
}
ComputeCloseTime(news, kvbase)
}

View File

@ -2,6 +2,7 @@ package grpc
import (
"context"
"github.com/zilliztech/milvus-distributed/internal/proto/schemapb"
"github.com/zilliztech/milvus-distributed/internal/proto/servicepb"
"net"
"strconv"
@ -11,11 +12,10 @@ import (
"github.com/zilliztech/milvus-distributed/internal/proto/commonpb"
"github.com/zilliztech/milvus-distributed/internal/proto/internalpb"
"github.com/zilliztech/milvus-distributed/internal/proto/masterpb"
messagepb "github.com/zilliztech/milvus-distributed/internal/proto/message"
"google.golang.org/grpc"
)
func Server(ch chan *messagepb.Mapping, errch chan error, kvbase kv.Base) {
func Server(ch chan *schemapb.CollectionSchema, errch chan error, kvbase kv.Base) {
defaultGRPCPort := ":"
defaultGRPCPort += strconv.FormatInt(int64(conf.Config.Master.Port), 10)
lis, err := net.Listen("tcp", defaultGRPCPort)
@ -34,7 +34,7 @@ func Server(ch chan *messagepb.Mapping, errch chan error, kvbase kv.Base) {
}
type GRPCMasterServer struct {
CreateRequest chan *messagepb.Mapping
CreateRequest chan *schemapb.CollectionSchema
kvbase kv.Base
}

View File

@ -8,7 +8,7 @@ import (
"github.com/zilliztech/milvus-distributed/internal/conf"
"github.com/zilliztech/milvus-distributed/internal/master/controller"
milvusgrpc "github.com/zilliztech/milvus-distributed/internal/master/grpc"
messagepb "github.com/zilliztech/milvus-distributed/internal/proto/message"
"github.com/zilliztech/milvus-distributed/internal/proto/schemapb"
"github.com/zilliztech/milvus-distributed/internal/master/kv"
"go.etcd.io/etcd/clientv3"
@ -16,7 +16,7 @@ import (
func Run() {
kvbase := newKvBase()
collectionChan := make(chan *messagepb.Mapping)
collectionChan := make(chan *schemapb.CollectionSchema)
defer close(collectionChan)
errorch := make(chan error)
@ -25,6 +25,7 @@ func Run() {
go milvusgrpc.Server(collectionChan, errorch, kvbase)
go controller.SegmentStatsController(kvbase, errorch)
go controller.CollectionController(collectionChan, kvbase, errorch)
//go timetick.TimeTickService()
for {
for v := range errorch {
log.Fatal(v)

View File

@ -0,0 +1,211 @@
package timesync
import (
"context"
"github.com/zilliztech/milvus-distributed/internal/conf"
"log"
"sort"
"strconv"
"sync"
"time"
"github.com/apache/pulsar-client-go/pulsar"
"github.com/golang/protobuf/proto"
pb "github.com/zilliztech/milvus-distributed/internal/proto/message"
)
const stopReadFlagId int64 = -1
type TimeTickReader struct {
pulsarClient pulsar.Client
timeTickConsumer pulsar.Consumer
readerProducer []pulsar.Producer
interval int64
proxyIdList []int64
timeTickPeerProxy map[int64]uint64
ctx context.Context
}
func (r *TimeTickReader) Start(){
go r.readTimeTick()
go r.timeSync()
}
func (r *TimeTickReader) Close() {
if r.timeTickConsumer != nil {
r.timeTickConsumer.Close()
}
for i := 0; i < len(r.readerProducer); i++ {
if r.readerProducer[i] != nil {
r.readerProducer[i].Close()
}
}
if r.pulsarClient != nil {
r.pulsarClient.Close()
}
}
func (r *TimeTickReader) timeSync() {
ctx := r.ctx
for {
select {
case <-ctx.Done():
return
default:
time.Sleep(time.Millisecond * time.Duration(r.interval))
var minTimeStamp uint64
for _, minTimeStamp = range r.timeTickPeerProxy {
break
}
for _, ts := range r.timeTickPeerProxy {
if ts < minTimeStamp {
minTimeStamp = ts
}
}
//send timestamp flag to reader channel
msg := pb.InsertOrDeleteMsg{Timestamp: minTimeStamp, ClientId: stopReadFlagId}
payload, err := proto.Marshal(&msg)
if err != nil {
//TODO log error
log.Printf("Marshal InsertOrDeleteMsg flag error %v", err)
} else {
wg := sync.WaitGroup{}
wg.Add(len(r.readerProducer))
for index := range r.readerProducer {
go r.sendEOFMsg(ctx, &pulsar.ProducerMessage{Payload: payload}, index, &wg)
}
wg.Wait()
}
}
}
}
func (r *TimeTickReader) readTimeTick() {
for {
select {
case <-r.ctx.Done():
return
case cm, ok := <-r.timeTickConsumer.Chan():
if ok == false {
log.Printf("timesync consumer closed")
}
msg := cm.Message
var tsm pb.TimeSyncMsg
if err := proto.Unmarshal(msg.Payload(), &tsm); err != nil {
log.Printf("UnMarshal timetick flag error %v", err)
}
r.timeTickPeerProxy[tsm.Peer_Id] = tsm.Timestamp
r.timeTickConsumer.AckID(msg.ID())
}
}
}
func (r *TimeTickReader) sendEOFMsg(ctx context.Context, msg *pulsar.ProducerMessage, index int, wg *sync.WaitGroup) {
if _, err := r.readerProducer[index].Send(ctx, msg); err != nil {
log.Printf("Send timesync flag error %v", err)
}
wg.Done()
}
func TimeTickService() {
timeTickTopic := "timeTick"
timeTickSubName := "master"
readTopics := make([]string, 0)
for i := conf.Config.Reader.TopicStart; i < conf.Config.Reader.TopicEnd; i++ {
str := "InsertOrDelete-"
str = str + strconv.Itoa(i)
readTopics = append(readTopics, str)
}
proxyIdList := conf.Config.Master.ProxyIdList
timeTickReader := newTimeTickReader(context.Background(), timeTickTopic, timeTickSubName, readTopics, proxyIdList)
timeTickReader.Start()
}
func newTimeTickReader(
ctx context.Context,
timeTickTopic string,
timeTickSubName string,
readTopics []string,
proxyIdList []int64,
) *TimeTickReader {
pulsarAddr := "pulsar://"
pulsarAddr += conf.Config.Pulsar.Address
pulsarAddr += ":"
pulsarAddr += strconv.FormatInt(int64(conf.Config.Pulsar.Port), 10)
interval := int64(conf.Config.Timesync.Interval)
//check if proxyId has duplication
if len(proxyIdList) == 0 {
log.Printf("proxy id list is empty")
}
if len(proxyIdList) > 1 {
sort.Slice(proxyIdList, func(i int, j int) bool { return proxyIdList[i] < proxyIdList[j] })
}
for i := 1; i < len(proxyIdList); i++ {
if proxyIdList[i] == proxyIdList[i-1] {
log.Printf("there are two proxies have the same id = %d", proxyIdList[i])
}
}
r := TimeTickReader{}
r.interval = interval
r.proxyIdList = proxyIdList
readerQueueSize := conf.Config.Reader.ReaderQueueSize
//check if read topic is empty
if len(readTopics) == 0 {
log.Printf("read topic is empyt")
}
//set default value
if readerQueueSize == 0 {
readerQueueSize = 1024
}
r.timeTickPeerProxy = make(map[int64]uint64)
r.ctx = ctx
var client pulsar.Client
var err error
if conf.Config.Pulsar.Authentication {
client, err = pulsar.NewClient(pulsar.ClientOptions{
URL: pulsarAddr,
Authentication: pulsar.NewAuthenticationToken(conf.Config.Pulsar.Token),
})
} else {
client, err = pulsar.NewClient(pulsar.ClientOptions{URL: pulsarAddr})
}
if err != nil {
log.Printf("connect pulsar failed, %v", err)
}
r.pulsarClient = client
timeSyncChan := make(chan pulsar.ConsumerMessage, len(r.proxyIdList))
if r.timeTickConsumer, err = r.pulsarClient.Subscribe(pulsar.ConsumerOptions{
Topic: timeTickTopic,
SubscriptionName: timeTickSubName,
Type: pulsar.KeyShared,
SubscriptionInitialPosition: pulsar.SubscriptionPositionEarliest,
MessageChannel: timeSyncChan,
}); err != nil {
log.Printf("failed to subscribe topic %s, error = %v", timeTickTopic, err)
}
r.readerProducer = make([]pulsar.Producer, 0, len(readTopics))
for i := 0; i < len(readTopics); i++ {
rp, err := r.pulsarClient.CreateProducer(pulsar.ProducerOptions{Topic: readTopics[i]})
if err != nil {
log.Printf("failed to create reader producer %s, error = %v", readTopics[i], err)
}
r.readerProducer = append(r.readerProducer, rp)
}
return &r
}

View File

@ -3,6 +3,8 @@ package proxy
import (
"github.com/apache/pulsar-client-go/pulsar"
pb "github.com/zilliztech/milvus-distributed/internal/proto/message"
"github.com/zilliztech/milvus-distributed/internal/proto/commonpb"
//pb "github.com/zilliztech/milvus-distributed/internal/proto/message"
"github.com/golang/protobuf/proto"
"log"
"sync"
@ -27,23 +29,23 @@ func (req *manipulationReq) Type() pb.ReqType {
return req.ReqType
}
func (req *manipulationReq) PreExecute() pb.Status {
return pb.Status{ErrorCode: pb.ErrorCode_SUCCESS}
func (req *manipulationReq) PreExecute() commonpb.Status {
return commonpb.Status{ErrorCode: commonpb.ErrorCode_SUCCESS}
}
func (req *manipulationReq) Execute() pb.Status {
func (req *manipulationReq) Execute() commonpb.Status {
req.proxy.reqSch.manipulationsChan <- req
return pb.Status{ErrorCode: pb.ErrorCode_SUCCESS}
return commonpb.Status{ErrorCode: commonpb.ErrorCode_SUCCESS}
}
func (req *manipulationReq) PostExecute() pb.Status { // send into pulsar
func (req *manipulationReq) PostExecute() commonpb.Status { // send into pulsar
req.wg.Add(1)
return pb.Status{ErrorCode: pb.ErrorCode_SUCCESS}
return commonpb.Status{ErrorCode: commonpb.ErrorCode_SUCCESS}
}
func (req *manipulationReq) WaitToFinish() pb.Status { // wait unitl send into pulsar
func (req *manipulationReq) WaitToFinish() commonpb.Status { // wait unitl send into pulsar
req.wg.Wait()
return pb.Status{ErrorCode: pb.ErrorCode_SUCCESS}
return commonpb.Status{ErrorCode: commonpb.ErrorCode_SUCCESS}
}
func (s *proxyServer) restartManipulationRoutine(buf_size int) error {
@ -77,7 +79,7 @@ func (s *proxyServer) restartManipulationRoutine(buf_size int) error {
return
case ip := <-s.reqSch.manipulationsChan:
ts, st := s.getTimestamp(1)
if st.ErrorCode != pb.ErrorCode_SUCCESS {
if st.ErrorCode != commonpb.ErrorCode_SUCCESS {
log.Printf("get time stamp failed, error code = %d, msg = %s, drop inset rows = %d", st.ErrorCode, st.Reason, len(ip.RowsData))
continue
}

View File

@ -5,17 +5,19 @@ import (
"fmt"
"github.com/apache/pulsar-client-go/pulsar"
"github.com/zilliztech/milvus-distributed/internal/conf"
pb "github.com/zilliztech/milvus-distributed/internal/proto/message"
"github.com/zilliztech/milvus-distributed/internal/proto/internalpb"
"github.com/zilliztech/milvus-distributed/internal/proto/commonpb"
etcd "go.etcd.io/etcd/clientv3"
"strconv"
)
type BaseRequest interface {
Type() pb.ReqType
PreExecute() pb.Status
Execute() pb.Status
PostExecute() pb.Status
WaitToFinish() pb.Status
Type() internalpb.ReqType
PreExecute() commonpb.Status
Execute() commonpb.Status
PostExecute() commonpb.Status
WaitToFinish() commonpb.Status
}
type ProxyOptions struct {
@ -156,13 +158,13 @@ func StartProxy(opt *ProxyOptions) error {
peer_id: opt.timeTickPeerId,
ctx: opt.ctx,
areRequestsDelivered: func(ts Timestamp) bool { return srv.reqSch.AreRequestsDelivered(ts, 2) },
getTimestamp: func() (Timestamp, pb.Status) {
getTimestamp: func() (Timestamp, commonpb.Status) {
ts, st := tso.GetTimestamp(1)
return ts[0], st
},
}
s := tt.Restart()
if s.ErrorCode != pb.ErrorCode_SUCCESS {
if s.ErrorCode != commonpb.ErrorCode_SUCCESS {
return fmt.Errorf(s.Reason)
}

View File

@ -5,7 +5,7 @@ import (
"encoding/binary"
"encoding/json"
"github.com/apache/pulsar-client-go/pulsar"
mpb "github.com/zilliztech/milvus-distributed/internal/proto/master"
"github.com/zilliztech/milvus-distributed/internal/proto/etcdpb"
pb "github.com/zilliztech/milvus-distributed/internal/proto/message"
"github.com/golang/protobuf/proto"
"github.com/stretchr/testify/assert"
@ -160,28 +160,24 @@ func TestProxyNode(t *testing.T) {
}
}()
cm100 := mpb.Collection{
cm100 := etcdpb.CollectionMeta{
Id: 100,
Name: "cm100",
Schema: nil,
CreateTime: 0,
SegmentIds: []uint64{101, 102},
PartitionTags: nil,
Indexes: nil,
}
sm101 := mpb.Segment{
sm101 := etcdpb.SegmentMeta{
SegmentId: 101,
CollectionId: 100,
ChannelStart: 0,
ChannelEnd: 1,
Status: mpb.SegmentStatus_OPENED,
}
sm102 := mpb.Segment{
sm102 := etcdpb.SegmentMeta{
SegmentId: 102,
CollectionId: 100,
ChannelStart: 1,
ChannelEnd: 2,
Status: mpb.SegmentStatus_OPENED,
}
if cm100b, err := json.Marshal(&cm100); err != nil {
t.Fatal(err)

View File

@ -5,7 +5,6 @@ import (
"github.com/golang/protobuf/proto"
"github.com/zilliztech/milvus-distributed/internal/proto/commonpb"
"github.com/zilliztech/milvus-distributed/internal/proto/internalpb"
pb "github.com/zilliztech/milvus-distributed/internal/proto/message"
"github.com/zilliztech/milvus-distributed/internal/proto/servicepb"
"log"
"sync"
@ -75,7 +74,7 @@ func (s *proxyServer) restartQueryRoutine(buf_size int) error {
return
case qm := <-s.reqSch.queryChan:
ts, st := s.getTimestamp(1)
if st.ErrorCode != pb.ErrorCode_SUCCESS {
if st.ErrorCode != commonpb.ErrorCode_SUCCESS {
log.Printf("get time stamp failed, error code = %d, msg = %s", st.ErrorCode, st.Reason)
break
}

View File

@ -4,24 +4,25 @@ import (
"context"
"encoding/json"
"fmt"
"log"
"net"
"sync"
"time"
"github.com/golang/protobuf/proto"
"github.com/zilliztech/milvus-distributed/internal/master/collection"
"github.com/zilliztech/milvus-distributed/internal/proto/commonpb"
mpb "github.com/zilliztech/milvus-distributed/internal/proto/masterpb"
"github.com/zilliztech/milvus-distributed/internal/proto/internalpb"
"github.com/zilliztech/milvus-distributed/internal/proto/etcdpb"
"github.com/zilliztech/milvus-distributed/internal/proto/internalpb"
mpb "github.com/zilliztech/milvus-distributed/internal/proto/masterpb"
pb "github.com/zilliztech/milvus-distributed/internal/proto/message"
"github.com/zilliztech/milvus-distributed/internal/proto/schemapb"
"github.com/zilliztech/milvus-distributed/internal/proto/servicepb"
pb "github.com/zilliztech/milvus-distributed/internal/proto/message"
"github.com/zilliztech/milvus-distributed/internal/master/collection"
"github.com/golang/protobuf/proto"
etcd "go.etcd.io/etcd/clientv3"
"go.uber.org/atomic"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
"log"
"net"
"sync"
"time"
)
const (
@ -42,7 +43,7 @@ type proxyServer struct {
resultGroup string
numReaderNode int
proxyId int64
getTimestamp func(count uint32) ([]Timestamp, pb.Status)
getTimestamp func(count uint32) ([]Timestamp, commonpb.Status)
client *etcd.Client
ctx context.Context
wg sync.WaitGroup
@ -59,8 +60,6 @@ type proxyServer struct {
queryId atomic.Uint64
}
func (s *proxyServer) CreateCollection(ctx context.Context, req *schemapb.CollectionSchema) (*commonpb.Status, error) {
return &commonpb.Status{
ErrorCode: 0,
@ -68,7 +67,6 @@ func (s *proxyServer) CreateCollection(ctx context.Context, req *schemapb.Collec
}, nil
}
func (s *proxyServer) DropCollection(ctx context.Context, req *servicepb.CollectionName) (*commonpb.Status, error) {
return &commonpb.Status{
ErrorCode: 0,
@ -83,7 +81,7 @@ func (s *proxyServer) HasCollection(ctx context.Context, req *servicepb.Collecti
Reason: "",
},
Value: true,
},nil
}, nil
}
func (s *proxyServer) DescribeCollection(ctx context.Context, req *servicepb.CollectionName) (*servicepb.CollectionDescription, error) {
@ -92,19 +90,18 @@ func (s *proxyServer) DescribeCollection(ctx context.Context, req *servicepb.Col
ErrorCode: 0,
Reason: "",
},
},nil
}, nil
}
func (s *proxyServer) ShowCollections(ctx context.Context, req * commonpb.Empty) (*servicepb.StringListResponse, error) {
func (s *proxyServer) ShowCollections(ctx context.Context, req *commonpb.Empty) (*servicepb.StringListResponse, error) {
return &servicepb.StringListResponse{
Status: &commonpb.Status{
ErrorCode: 0,
Reason: "",
},
},nil
}, nil
}
func (s *proxyServer) CreatePartition(ctx context.Context, in *servicepb.PartitionName) (*commonpb.Status, error) {
return &commonpb.Status{
ErrorCode: 0,
@ -126,7 +123,7 @@ func (s *proxyServer) HasPartition(ctx context.Context, in *servicepb.PartitionN
Reason: "",
},
Value: true,
},nil
}, nil
}
func (s *proxyServer) DescribePartition(ctx context.Context, in *servicepb.PartitionName) (*servicepb.PartitionDescription, error) {
@ -135,7 +132,7 @@ func (s *proxyServer) DescribePartition(ctx context.Context, in *servicepb.Parti
ErrorCode: 0,
Reason: "",
},
},nil
}, nil
}
func (s *proxyServer) ShowPartitions(ctx context.Context, req *servicepb.CollectionName) (*servicepb.StringListResponse, error) {
@ -144,10 +141,10 @@ func (s *proxyServer) ShowPartitions(ctx context.Context, req *servicepb.Collect
ErrorCode: 0,
Reason: "",
},
},nil
}, nil
}
func (s *proxyServer) DeleteByID(ctx context.Context, req *pb.DeleteByIDParam) (*pb.Status, error) {
func (s *proxyServer) DeleteByID(ctx context.Context, req *pb.DeleteByIDParam) (*commonpb.Status, error) {
log.Printf("delete entites, total = %d", len(req.IdArray))
pm := &manipulationReq{
ManipulationReqMsg: pb.ManipulationReqMsg{
@ -161,23 +158,22 @@ func (s *proxyServer) DeleteByID(ctx context.Context, req *pb.DeleteByIDParam) (
pm.PrimaryKeys = append(pm.PrimaryKeys, uint64(id))
}
if len(pm.PrimaryKeys) > 1 {
if st := pm.PreExecute(); st.ErrorCode != pb.ErrorCode_SUCCESS {
if st := pm.PreExecute(); st.ErrorCode != commonpb.ErrorCode_SUCCESS {
return &st, nil
}
if st := pm.Execute(); st.ErrorCode != pb.ErrorCode_SUCCESS {
if st := pm.Execute(); st.ErrorCode != commonpb.ErrorCode_SUCCESS {
return &st, nil
}
if st := pm.PostExecute(); st.ErrorCode != pb.ErrorCode_SUCCESS {
if st := pm.PostExecute(); st.ErrorCode != commonpb.ErrorCode_SUCCESS {
return &st, nil
}
if st := pm.WaitToFinish(); st.ErrorCode != pb.ErrorCode_SUCCESS {
if st := pm.WaitToFinish(); st.ErrorCode != commonpb.ErrorCode_SUCCESS {
return &st, nil
}
}
return &pb.Status{ErrorCode: pb.ErrorCode_SUCCESS}, nil
return &commonpb.Status{ErrorCode: commonpb.ErrorCode_SUCCESS}, nil
}
func (s *proxyServer) Insert(ctx context.Context, req *servicepb.RowBatch) (*servicepb.IntegerRangeResponse, error) {
log.Printf("Insert Entities, total = %d", len(req.RowData))
ipm := make(map[uint32]*manipulationReq)
@ -223,27 +219,27 @@ func (s *proxyServer) Insert(ctx context.Context, req *servicepb.RowBatch) (*ser
ip = ipm[hash]
}
ip.PrimaryKeys = append(ip.PrimaryKeys, key)
ip.RowsData = append(ip.RowsData, &pb.RowData{Blob:req.RowData[i].Value}) // czs_tag
ip.RowsData = append(ip.RowsData, &pb.RowData{Blob: req.RowData[i].Value}) // czs_tag
}
for _, ip := range ipm {
if st := ip.PreExecute(); st.ErrorCode != pb.ErrorCode_SUCCESS { //do nothing
if st := ip.PreExecute(); st.ErrorCode != commonpb.ErrorCode_SUCCESS { //do nothing
return &servicepb.IntegerRangeResponse{
Status: &commonpb.Status{ErrorCode: commonpb.ErrorCode_UNEXPECTED_ERROR},
Status: &commonpb.Status{ErrorCode: commonpb.ErrorCode_UNEXPECTED_ERROR},
}, nil
}
if st := ip.Execute(); st.ErrorCode != pb.ErrorCode_SUCCESS { // push into chan
if st := ip.Execute(); st.ErrorCode != commonpb.ErrorCode_SUCCESS { // push into chan
return &servicepb.IntegerRangeResponse{
Status: &commonpb.Status{ErrorCode: commonpb.ErrorCode_UNEXPECTED_ERROR},
Status: &commonpb.Status{ErrorCode: commonpb.ErrorCode_UNEXPECTED_ERROR},
}, nil
}
if st := ip.PostExecute(); st.ErrorCode != pb.ErrorCode_SUCCESS { //post to pulsar
if st := ip.PostExecute(); st.ErrorCode != commonpb.ErrorCode_SUCCESS { //post to pulsar
return &servicepb.IntegerRangeResponse{
Status: &commonpb.Status{ErrorCode: commonpb.ErrorCode_UNEXPECTED_ERROR},
Status: &commonpb.Status{ErrorCode: commonpb.ErrorCode_UNEXPECTED_ERROR},
}, nil
}
}
for _, ip := range ipm {
if st := ip.WaitToFinish(); st.ErrorCode != pb.ErrorCode_SUCCESS {
if st := ip.WaitToFinish(); st.ErrorCode != commonpb.ErrorCode_SUCCESS {
log.Printf("Wait to finish failed, error code = %d", st.ErrorCode)
}
}
@ -258,10 +254,10 @@ func (s *proxyServer) Insert(ctx context.Context, req *servicepb.RowBatch) (*ser
func (s *proxyServer) Search(ctx context.Context, req *servicepb.Query) (*servicepb.QueryResult, error) {
qm := &queryReq{
SearchRequest: internalpb.SearchRequest{
ReqType: internalpb.ReqType_kSearch,
ProxyId: s.proxyId,
ReqId: s.queryId.Add(1),
Timestamp: 0,
ReqType: internalpb.ReqType_kSearch,
ProxyId: s.proxyId,
ReqId: s.queryId.Add(1),
Timestamp: 0,
ResultChannelId: 0,
},
proxy: s,
@ -269,22 +265,22 @@ func (s *proxyServer) Search(ctx context.Context, req *servicepb.Query) (*servic
log.Printf("search on collection %s, proxy id = %d, query id = %d", req.CollectionName, qm.ProxyId, qm.ReqId)
if st := qm.PreExecute(); st.ErrorCode != commonpb.ErrorCode_SUCCESS {
return &servicepb.QueryResult{
Status: &st,
Status: &st,
}, nil
}
if st := qm.Execute(); st.ErrorCode != commonpb.ErrorCode_SUCCESS {
return &servicepb.QueryResult{
Status: &st,
Status: &st,
}, nil
}
if st := qm.PostExecute(); st.ErrorCode != commonpb.ErrorCode_SUCCESS {
return &servicepb.QueryResult{
Status: &st,
Status: &st,
}, nil
}
if st := qm.WaitToFinish(); st.ErrorCode != commonpb.ErrorCode_SUCCESS {
return &servicepb.QueryResult{
Status: &st,
Status: &st,
}, nil
}
return s.reduceResults(qm), nil

View File

@ -5,8 +5,10 @@ import (
"encoding/binary"
"encoding/json"
"github.com/apache/pulsar-client-go/pulsar"
mpb "github.com/zilliztech/milvus-distributed/internal/proto/master"
mpb "github.com/zilliztech/milvus-distributed/internal/proto/masterpb"
"github.com/zilliztech/milvus-distributed/internal/proto/etcdpb"
pb "github.com/zilliztech/milvus-distributed/internal/proto/message"
"github.com/zilliztech/milvus-distributed/internal/proto/commonpb"
"github.com/golang/protobuf/proto"
"github.com/stretchr/testify/assert"
"go.etcd.io/etcd/clientv3"
@ -61,13 +63,13 @@ func startTestProxyServer(proxy_addr string, master_addr string, t *testing.T) *
resultGroup: "reusltG",
numReaderNode: 2,
proxyId: 1,
getTimestamp: func(count uint32) ([]Timestamp, pb.Status) {
getTimestamp: func(count uint32) ([]Timestamp, commonpb.Status) {
timestamp += 100
t := make([]Timestamp, count)
for i := 0; i < int(count); i++ {
t[i] = Timestamp(timestamp)
}
return t, pb.Status{ErrorCode: pb.ErrorCode_SUCCESS}
return t, commonpb.Status{ErrorCode: commonpb.ErrorCode_SUCCESS}
},
client: client,
ctx: ctx,
@ -121,18 +123,17 @@ func TestProxyServer_WatchEtcd(t *testing.T) {
defer client.Close()
ctx, _ := context.WithTimeout(context.Background(), 2*time.Second)
col1 := mpb.Collection{
col1 := etcdpb.CollectionMeta{
Id: 1,
Name: "c1",
SegmentIds: []uint64{2, 3},
}
seg2 := mpb.Segment{
seg2 := etcdpb.SegmentMeta{
SegmentId: 2,
Rows: 10,
NumRows: 10,
}
seg3 := mpb.Segment{
seg3 := etcdpb.SegmentMeta{
SegmentId: 3,
Rows: 10,
NumRows: 10,
}
if cb1, err := json.Marshal(&col1); err != nil {
t.Fatal(err)
@ -171,14 +172,13 @@ func TestProxyServer_WatchEtcd(t *testing.T) {
assert.Equalf(t, cr.Status.ErrorCode, pb.ErrorCode_SUCCESS, "CountCollection failed : %s", cr.Status.Reason)
assert.Equalf(t, cr.CollectionRowCount, int64(20), "collection count expect to be 20, count = %d", cr.CollectionRowCount)
col4 := mpb.Collection{
col4 := etcdpb.CollectionMeta{
Id: 4,
Name: "c4",
SegmentIds: []uint64{5},
}
seg5 := mpb.Segment{
seg5 := etcdpb.SegmentMeta{
SegmentId: 5,
Rows: 10,
NumRows: 10,
}
if cb4, err := json.Marshal(&col4); err != nil {
t.Fatal(err)
@ -202,28 +202,24 @@ func TestProxyServer_InsertAndDelete(t *testing.T) {
defer client.Close()
ctx, _ := context.WithTimeout(context.Background(), 2*time.Second)
col10 := mpb.Collection{
col10 := etcdpb.CollectionMeta{
Id: 10,
Name: "col10",
Schema: nil,
CreateTime: 0,
SegmentIds: []uint64{11, 12},
PartitionTags: nil,
Indexes: nil,
}
seg11 := mpb.Segment{
seg11 := etcdpb.SegmentMeta{
SegmentId: 11,
CollectionId: 10,
ChannelStart: 0,
ChannelEnd: 1,
Status: mpb.SegmentStatus_OPENED,
}
seg12 := mpb.Segment{
seg12 := etcdpb.SegmentMeta{
SegmentId: 12,
CollectionId: 10,
ChannelStart: 1,
ChannelEnd: 2,
Status: mpb.SegmentStatus_OPENED,
}
if cb10, err := json.Marshal(&col10); err != nil {
t.Fatal(err)

View File

@ -3,7 +3,7 @@ package proxy
import (
"context"
"fmt"
pb "github.com/zilliztech/milvus-distributed/internal/proto/message"
"github.com/zilliztech/milvus-distributed/internal/proto/commonpb"
etcd "go.etcd.io/etcd/clientv3"
"log"
"strconv"
@ -51,7 +51,7 @@ func (tso *timestampOracle) Restart(id int64) {
select {
case <-ticker:
_, s := tso.GetTimestamp(1)
if s.ErrorCode == pb.ErrorCode_SUCCESS {
if s.ErrorCode == commonpb.ErrorCode_SUCCESS {
_ = tso.saveTimestamp()
}
break
@ -65,7 +65,7 @@ func (tso *timestampOracle) Restart(id int64) {
}()
}
func (tso *timestampOracle) GetTimestamp(count uint32) ([]Timestamp, pb.Status) {
func (tso *timestampOracle) GetTimestamp(count uint32) ([]Timestamp, commonpb.Status) {
physical := uint64(time.Now().UnixNano()) / uint64(1e6)
var ctso timestamp
tso.mux.Lock()
@ -80,31 +80,31 @@ func (tso *timestampOracle) GetTimestamp(count uint32) ([]Timestamp, pb.Status)
ctso.logical = uint64(i)
tt = append(tt, ToTimeStamp(&ctso))
}
return tt, pb.Status{ErrorCode: pb.ErrorCode_SUCCESS}
return tt, commonpb.Status{ErrorCode: commonpb.ErrorCode_SUCCESS}
}
func (tso *timestampOracle) saveTimestamp() pb.Status {
func (tso *timestampOracle) saveTimestamp() commonpb.Status {
tso.mux.Lock()
physical := tso.tso.physical
tso.mux.Unlock()
if _, err := tso.client.Put(tso.ctx, tso.rootPath+tsoKeyPath, strconv.FormatUint(physical, 10)); err != nil {
return pb.Status{ErrorCode: pb.ErrorCode_UNEXPECTED_ERROR, Reason: fmt.Sprintf("put into etcd failed, error = %v", err)}
return commonpb.Status{ErrorCode: commonpb.ErrorCode_UNEXPECTED_ERROR, Reason: fmt.Sprintf("put into etcd failed, error = %v", err)}
}
tso.mux.Lock()
tso.lastSavedTime = physical
tso.mux.Unlock()
return pb.Status{ErrorCode: pb.ErrorCode_SUCCESS}
return commonpb.Status{ErrorCode: commonpb.ErrorCode_SUCCESS}
}
func (tso *timestampOracle) loadTimestamp() pb.Status {
func (tso *timestampOracle) loadTimestamp() commonpb.Status {
ts, err := tso.client.Get(tso.ctx, tso.rootPath+tsoKeyPath)
if err != nil {
return pb.Status{ErrorCode: pb.ErrorCode_UNEXPECTED_ERROR, Reason: fmt.Sprintf("get from etcd failed, error = %v", err)}
return commonpb.Status{ErrorCode: commonpb.ErrorCode_UNEXPECTED_ERROR, Reason: fmt.Sprintf("get from etcd failed, error = %v", err)}
}
if len(ts.Kvs) != 0 {
n, err := strconv.ParseUint(string(ts.Kvs[0].Value), 10, 64)
if err != nil {
return pb.Status{ErrorCode: pb.ErrorCode_UNEXPECTED_ERROR, Reason: fmt.Sprintf("ParseUint failed, error = %v", err)}
return commonpb.Status{ErrorCode: commonpb.ErrorCode_UNEXPECTED_ERROR, Reason: fmt.Sprintf("ParseUint failed, error = %v", err)}
}
tso.mux.Lock()
tso.tso.physical = n
@ -116,5 +116,5 @@ func (tso *timestampOracle) loadTimestamp() pb.Status {
tso.lastSavedTime = tso.tso.physical
tso.mux.Unlock()
}
return pb.Status{ErrorCode: pb.ErrorCode_UNEXPECTED_ERROR}
return commonpb.Status{ErrorCode: commonpb.ErrorCode_UNEXPECTED_ERROR}
}

View File

@ -4,6 +4,7 @@ import (
"context"
"fmt"
"github.com/apache/pulsar-client-go/pulsar"
"github.com/zilliztech/milvus-distributed/internal/proto/commonpb"
pb "github.com/zilliztech/milvus-distributed/internal/proto/message"
"github.com/golang/protobuf/proto"
"log"
@ -18,19 +19,19 @@ type timeTick struct {
peer_id int64
ctx context.Context
areRequestsDelivered func(ts Timestamp) bool
getTimestamp func() (Timestamp, pb.Status)
getTimestamp func() (Timestamp, commonpb.Status)
}
func (tt *timeTick) tick() pb.Status {
func (tt *timeTick) tick() commonpb.Status {
if tt.lastTick == tt.currentTick {
ts, s := tt.getTimestamp()
if s.ErrorCode != pb.ErrorCode_SUCCESS {
if s.ErrorCode != commonpb.ErrorCode_SUCCESS {
return s
}
tt.currentTick = ts
}
if tt.areRequestsDelivered(tt.currentTick) == false {
return pb.Status{ErrorCode: pb.ErrorCode_SUCCESS}
return commonpb.Status{ErrorCode: commonpb.ErrorCode_SUCCESS}
}
tsm := pb.TimeSyncMsg{
Timestamp: uint64(tt.currentTick),
@ -39,19 +40,19 @@ func (tt *timeTick) tick() pb.Status {
}
payload, err := proto.Marshal(&tsm)
if err != nil {
return pb.Status{ErrorCode: pb.ErrorCode_UNEXPECTED_ERROR, Reason: fmt.Sprintf("marshal TimeSync failed, error = %v", err)}
return commonpb.Status{ErrorCode: commonpb.ErrorCode_UNEXPECTED_ERROR, Reason: fmt.Sprintf("marshal TimeSync failed, error = %v", err)}
}
if _, err := tt.pulsarProducer.Send(tt.ctx, &pulsar.ProducerMessage{Payload: payload}); err != nil {
return pb.Status{ErrorCode: pb.ErrorCode_UNEXPECTED_ERROR, Reason: fmt.Sprintf("send into pulsar failed, error = %v", err)}
return commonpb.Status{ErrorCode: commonpb.ErrorCode_UNEXPECTED_ERROR, Reason: fmt.Sprintf("send into pulsar failed, error = %v", err)}
}
tt.lastTick = tt.currentTick
return pb.Status{ErrorCode: pb.ErrorCode_SUCCESS}
return commonpb.Status{ErrorCode: commonpb.ErrorCode_SUCCESS}
}
func (tt *timeTick) Restart() pb.Status {
func (tt *timeTick) Restart() commonpb.Status {
tt.lastTick = 0
ts, s := tt.getTimestamp()
if s.ErrorCode != pb.ErrorCode_SUCCESS {
if s.ErrorCode != commonpb.ErrorCode_SUCCESS {
return s
}
tt.currentTick = ts
@ -61,7 +62,7 @@ func (tt *timeTick) Restart() pb.Status {
for {
select {
case <-tick:
if s := tt.tick(); s.ErrorCode != pb.ErrorCode_SUCCESS {
if s := tt.tick(); s.ErrorCode != commonpb.ErrorCode_SUCCESS {
log.Printf("timeTick error ,status = %d", int(s.ErrorCode))
}
case <-tt.ctx.Done():
@ -70,5 +71,5 @@ func (tt *timeTick) Restart() pb.Status {
}
}
}()
return pb.Status{ErrorCode: pb.ErrorCode_SUCCESS}
return commonpb.Status{ErrorCode: commonpb.ErrorCode_SUCCESS}
}

View File

@ -4,6 +4,7 @@ import (
"context"
"github.com/apache/pulsar-client-go/pulsar"
pb "github.com/zilliztech/milvus-distributed/internal/proto/message"
"github.com/zilliztech/milvus-distributed/internal/proto/commonpb"
"github.com/golang/protobuf/proto"
"github.com/stretchr/testify/assert"
"testing"
@ -35,9 +36,9 @@ func TestTimeTick(t *testing.T) {
peer_id: 1,
ctx: ctx,
areRequestsDelivered: func(ts Timestamp) bool { return true },
getTimestamp: func() (Timestamp, pb.Status) {
getTimestamp: func() (Timestamp, commonpb.Status) {
curTs = curTs + 100
return curTs, pb.Status{ErrorCode: pb.ErrorCode_SUCCESS}
return curTs, commonpb.Status{ErrorCode: commonpb.ErrorCode_SUCCESS}
},
}
tt.Restart()

View File

@ -19,7 +19,7 @@ CPP_BUILD_DIR="${CPP_SRC_DIR}/cmake_build"
BUILD_OUTPUT_DIR=${CPP_BUILD_DIR}
BUILD_TYPE="Release"
BUILD_UNITTEST="OFF"
INSTALL_PREFIX="${CPP_SRC_DIR}/milvus"
INSTALL_PREFIX="${CPP_SRC_DIR}/output"
MAKE_CLEAN="OFF"
BUILD_COVERAGE="OFF"
DB_PATH="/tmp/milvus"