admin/milvus
1
0
Fork 0
mirror of https://gitee.com/milvus-io/milvus.git synced 2025-12-28 22:45:26 +08:00
Code Issues Packages Projects Releases Wiki Activity

Read config from file in master

Browse Source 
Signed-off-by: rain <boyan.wang@zilliz.com>
This commit is contained in:
rain 2020-09-22 15:26:38 +08:00 committed by yefu.chen
parent 7812eed091
commit bb03b9bfde
4 changed files with 31 additions and 20 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

18
conf/conf.go
View File

@ -1,18 +1,24 @@
package conf
import (
"github.com/czs007/suvlim/storage/pkg/types"
yaml "gopkg.in/yaml.v2"
"io/ioutil"
"path"
"runtime"
"github.com/czs007/suvlim/storage/pkg/types"
yaml "gopkg.in/yaml.v2"
)
// yaml.MapSlice
type MasterConfig struct {
Address string
Port int32
PulsarURL string
PulsarMoniterInterval int32
PulsarTopic string
EtcdRootPath string
SegmentThreshole float32
DefaultGRPCPort string
EtcdEndPoints []string
}
type EtcdConfig struct {
@ -61,8 +67,8 @@ func init() {
}
func getCurrentFileDir() string {
_, fpath, _, _ := runtime.Caller(0)
return path.Dir(fpath)
_, fpath, _, _ := runtime.Caller(0)
return path.Dir(fpath)
}
func load_config() {

9
conf/config.yaml
View File

@ -10,8 +10,13 @@
# or implied. See the License for the specific language governing permissions and limitations under the License.
master:
address: localhost
port: 53100
pulsarurl: "pulsar://localhost:6650"
pulsarmoniterinterval: 1
pulsartopic: "monitor-topic"
etcdrootpath: "by-dev"
segmentthreshole: 10000
defaultgrpcport: ":53100"
etcdendpoints: ["127.0.0.1:12379"]
etcd:
address: localhost

2
pkg/master/README.md
View File

@ -21,4 +21,4 @@ go run cmd/master.go
### example
if master create a collection with uuid ```46e468ee-b34a-419d-85ed-80c56bfa4e90```
the corresponding key in etcd is /$(ETCD_ROOT_PATH)/collection/46e468ee-b34a-419d-85ed-80c56bfa4e90
the corresponding key in etcd is $(ETCD_ROOT_PATH)/collection/46e468ee-b34a-419d-85ed-80c56bfa4e90

22
pkg/master/server.go
View File

@ -8,7 +8,7 @@ import (
"strconv"
"time"
"github.com/czs007/suvlim/pkg/master/common"
"github.com/czs007/suvlim/conf"
pb "github.com/czs007/suvlim/pkg/master/grpc/master"
"github.com/czs007/suvlim/pkg/master/grpc/message"
messagepb "github.com/czs007/suvlim/pkg/master/grpc/message"
@ -33,11 +33,11 @@ func Run() {
func SegmentStatsController() {
cli, _ := clientv3.New(clientv3.Config{
Endpoints: []string{"127.0.0.1:12379"},
Endpoints: conf.Config.Master.EtcdEndPoints,
DialTimeout: 5 * time.Second,
})
defer cli.Close()
kvbase := kv.NewEtcdKVBase(cli, common.ETCD_ROOT_PATH)
kvbase := kv.NewEtcdKVBase(cli, conf.Config.Master.EtcdRootPath)
ssChan := make(chan mock.SegmentStats, 10)
defer close(ssChan)
@ -56,13 +56,13 @@ func SegmentStatsController() {
}
func ComputeCloseTime(ss mock.SegmentStats, kvbase kv.Base) error {
if int(ss.MemorySize) > common.SEGMENT_THRESHOLE*0.8 {
if int(ss.MemorySize) > int(conf.Config.Master.SegmentThreshole*0.8) {
currentTime := time.Now()
memRate := int(ss.MemoryRate)
if memRate == 0 {
memRate = 1
}
sec := common.SEGMENT_THRESHOLE * 0.2 / memRate
sec := int(conf.Config.Master.SegmentThreshole*0.2) / memRate
data, err := kvbase.Load("segment/" + strconv.Itoa(int(ss.SegementID)))
if err != nil {
return err
@ -138,7 +138,7 @@ func UpdateSegmentStatus(ss mock.SegmentStats, kvbase kv.Base) error {
}
func GRPCServer(ch chan *messagepb.Mapping) error {
lis, err := net.Listen("tcp", common.DEFAULT_GRPC_PORT)
lis, err := net.Listen("tcp", conf.Config.Master.DefaultGRPCPort)
if err != nil {
return err
}
@ -194,11 +194,11 @@ func (ms GRPCMasterServer) CreateIndex(ctx context.Context, in *messagepb.IndexP
func CollectionController(ch chan *messagepb.Mapping) {
cli, _ := clientv3.New(clientv3.Config{
Endpoints: []string{"127.0.0.1:12379"},
Endpoints: conf.Config.Master.EtcdEndPoints,
DialTimeout: 5 * time.Second,
})
defer cli.Close()
kvbase := kv.NewEtcdKVBase(cli, common.ETCD_ROOT_PATH)
kvbase := kv.NewEtcdKVBase(cli, conf.Config.Master.EtcdRootPath)
for collection := range ch {
sID := id.New().Uint64()
cID := id.New().Uint64()
@ -243,7 +243,7 @@ func WriteCollection2Datastore(collection *messagepb.Mapping) error {
DialTimeout: 5 * time.Second,
})
defer cli.Close()
kvbase := kv.NewEtcdKVBase(cli, common.ETCD_ROOT_PATH)
kvbase := kv.NewEtcdKVBase(cli, conf.Config.Master.EtcdRootPath)
sID := id.New().Uint64()
cID := id.New().Uint64()
fieldMetas := []*messagepb.FieldMeta{}
@ -281,11 +281,11 @@ func WriteCollection2Datastore(collection *messagepb.Mapping) error {
func UpdateCollectionIndex(index *messagepb.IndexParam) error {
cli, _ := clientv3.New(clientv3.Config{
Endpoints: []string{"127.0.0.1:12379"},
Endpoints: conf.Config.Master.EtcdEndPoints,
DialTimeout: 5 * time.Second,
})
defer cli.Close()
kvbase := kv.NewEtcdKVBase(cli, common.ETCD_ROOT_PATH)
kvbase := kv.NewEtcdKVBase(cli, conf.Config.Master.EtcdRootPath)
collectionName := index.CollectionName
c, err := kvbase.Load("collection/" + collectionName)
if err != nil {