milvus/internal/datacoord/param.go

243 lines
5.5 KiB
Go
Raw Normal View History

// 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.
package datacoord
import (
"path"
"strconv"
2021-06-11 22:04:41 +08:00
"strings"
"sync"
"github.com/milvus-io/milvus/internal/log"
"github.com/milvus-io/milvus/internal/util/paramtable"
)
type ParamTable struct {
paramtable.BaseTable
NodeID int64
IP string
Port int
// --- ETCD ---
2021-06-11 22:04:41 +08:00
EtcdEndpoints []string
MetaRootPath string
KvRootPath string
SegmentBinlogSubPath string
CollectionBinlogSubPath string
// --- Pulsar ---
PulsarAddress string
FlushStreamPosSubPath string
StatsStreamPosSubPath string
// segment
SegmentMaxSize float64
SegmentSealProportion float64
SegAssignmentExpiration int64
InsertChannelPrefixName string
StatisticsChannelName string
TimeTickChannelName string
SegmentInfoChannelName string
DataCoordSubscriptionName string
Fix bugs (#5676) * Remove redundant session startup Signed-off-by: sunby <bingyi.sun@zilliz.com> * Register datanode after start success Signed-off-by: sunby <bingyi.sun@zilliz.com> * fix meta snap shot Signed-off-by: yefu.chen <yefu.chen@zilliz.com> * fix datanode message stream channel Signed-off-by: yangxuan <xuan.yang@zilliz.com> * Fix bugs when drop empty collection Signed-off-by: sunby <bingyi.sun@zilliz.com> * Fix bug of getting pchan statistics from task scheduler Signed-off-by: dragondriver <jiquan.long@zilliz.com> * Fix i/dist/dataservice test code Signed-off-by: Congqi Xia <congqi.xia@zilliz.com> * Fix epoch lifetime not applied Signed-off-by: Congqi Xia <congqi.xia@zilliz.com> * fix datanode flowgraph dd node Signed-off-by: yangxuan <xuan.yang@zilliz.com> * Fix handle datanode timetick bug Signed-off-by: sunby <bingyi.sun@zilliz.com> * Remove repack function of dml stream Signed-off-by: dragondriver <jiquan.long@zilliz.com> * fix proxynode Signed-off-by: yefu.chen <yefu.chen@zilliz.com> * Apply extended seal policy Signed-off-by: Congqi Xia <congqi.xia@zilliz.com> * add check for time tick Signed-off-by: yefu.chen <yefu.chen@zilliz.com> * fix check Signed-off-by: yefu.chen <yefu.chen@zilliz.com> * Fix the repack function of dml stream Signed-off-by: dragondriver <jiquan.long@zilliz.com> * Fix the bug when send statistics of pchan Signed-off-by: dragondriver <jiquan.long@zilliz.com> * Fix the repack function when craete dml stream Signed-off-by: dragondriver <jiquan.long@zilliz.com> * fix bugs Signed-off-by: yefu.chen <yefu.chen@zilliz.com> * fix describe collection Signed-off-by: yefu.chen <yefu.chen@zilliz.com> * Fix bug when send timestamp statistics Signed-off-by: dragondriver <jiquan.long@zilliz.com> * fix data node Signed-off-by: yefu.chen <yefu.chen@zilliz.com> * Add length check before flush request Signed-off-by: Congqi Xia <congqi.xia@zilliz.com> * add log for data node Signed-off-by: yefu.chen <yefu.chen@zilliz.com> * Fix SaveBinlog bugs Signed-off-by: sunby <bingyi.sun@zilliz.com> * Add more log in datanode Signed-off-by: yangxuan <xuan.yang@zilliz.com> * Put SegmentState.Flushing as the last one in enum to fit the client Signed-off-by: sunby <bingyi.sun@zilliz.com> * Fix params in GetInsertBinlogPaths Signed-off-by: sunby <bingyi.sun@zilliz.com> * Rename policy Signed-off-by: sunby <bingyi.sun@zilliz.com> * Remove unused ddl functions and fields Signed-off-by: Congqi Xia <congqi.xia@zilliz.com> * Remove pchan when drop collection Signed-off-by: dragondriver <jiquan.long@zilliz.com> * Add balanced assignment policy Signed-off-by: sunby <bingyi.sun@zilliz.com> * fix master ut Signed-off-by: yefu.chen <yefu.chen@zilliz.com> * Add lock in session manager Signed-off-by: Congqi Xia <congqi.xia@zilliz.com> * add log for debug Signed-off-by: yefu.chen <yefu.chen@zilliz.com> * Fix some logic bug and typo Signed-off-by: Congqi Xia <congqi.xia@zilliz.com> * Fix recover bugs Signed-off-by: sunby <bingyi.sun@zilliz.com> * Get collection scheme of a specific timestamp Signed-off-by: yangxuan <xuan.yang@zilliz.com> * Change CheckPoint to SegmentInfo in VchannelInfo Signed-off-by: sunby <bingyi.sun@zilliz.com> * Recover Unflushed segment numOfRows Signed-off-by: yangxuan <xuan.yang@zilliz.com> * Fix dataservice unit tests Signed-off-by: sunby <bingyi.sun@zilliz.com> Co-authored-by: yefu.chen <yefu.chen@zilliz.com> Co-authored-by: yangxuan <xuan.yang@zilliz.com> Co-authored-by: dragondriver <jiquan.long@zilliz.com> Co-authored-by: Congqi Xia <congqi.xia@zilliz.com>
2021-06-08 19:25:37 +08:00
Log log.Config
}
var Params ParamTable
var once sync.Once
func (p *ParamTable) Init() {
once.Do(func() {
// load yaml
p.BaseTable.Init()
if err := p.LoadYaml("advanced/data_coord.yaml"); err != nil {
panic(err)
}
// set members
2021-06-11 22:04:41 +08:00
p.initEtcdEndpoints()
p.initMetaRootPath()
p.initKvRootPath()
p.initSegmentBinlogSubPath()
p.initCollectionBinlogSubPath()
p.initPulsarAddress()
p.initSegmentMaxSize()
p.initSegmentSealProportion()
p.initSegAssignmentExpiration()
p.initInsertChannelPrefixName()
p.initStatisticsChannelName()
p.initTimeTickChannelName()
p.initSegmentInfoChannelName()
p.initDataCoordSubscriptionName()
p.initLogCfg()
p.initFlushStreamPosSubPath()
p.initStatsStreamPosSubPath()
})
}
2021-06-11 22:04:41 +08:00
func (p *ParamTable) initEtcdEndpoints() {
endpoints, err := p.Load("_EtcdEndpoints")
if err != nil {
panic(err)
}
2021-06-11 22:04:41 +08:00
p.EtcdEndpoints = strings.Split(endpoints, ",")
}
func (p *ParamTable) initPulsarAddress() {
addr, err := p.Load("_PulsarAddress")
if err != nil {
panic(err)
}
p.PulsarAddress = addr
}
func (p *ParamTable) initMetaRootPath() {
rootPath, err := p.Load("etcd.rootPath")
if err != nil {
panic(err)
}
subPath, err := p.Load("etcd.metaSubPath")
if err != nil {
panic(err)
}
p.MetaRootPath = rootPath + "/" + subPath
}
func (p *ParamTable) initKvRootPath() {
rootPath, err := p.Load("etcd.rootPath")
if err != nil {
panic(err)
}
subPath, err := p.Load("etcd.kvSubPath")
if err != nil {
panic(err)
}
p.KvRootPath = rootPath + "/" + subPath
}
func (p *ParamTable) initSegmentBinlogSubPath() {
subPath, err := p.Load("etcd.segmentBinlogSubPath")
if err != nil {
panic(err)
}
p.SegmentBinlogSubPath = subPath
}
func (p *ParamTable) initCollectionBinlogSubPath() {
subPath, err := p.Load("etcd.collectionBinlogSubPath")
if err != nil {
panic(err)
}
p.CollectionBinlogSubPath = subPath
}
func (p *ParamTable) initSegmentMaxSize() {
p.SegmentMaxSize = p.ParseFloat("datacoord.segment.maxSize")
}
func (p *ParamTable) initSegmentSealProportion() {
p.SegmentSealProportion = p.ParseFloat("datacoord.segment.sealProportion")
}
func (p *ParamTable) initSegAssignmentExpiration() {
p.SegAssignmentExpiration = p.ParseInt64("datacoord.segment.assignmentExpiration")
}
func (p *ParamTable) initInsertChannelPrefixName() {
var err error
p.InsertChannelPrefixName, err = p.Load("msgChannel.chanNamePrefix.dataCoordInsertChannel")
if err != nil {
panic(err)
}
}
func (p *ParamTable) initStatisticsChannelName() {
var err error
p.StatisticsChannelName, err = p.Load("msgChannel.chanNamePrefix.dataCoordStatistic")
if err != nil {
panic(err)
}
}
func (p *ParamTable) initTimeTickChannelName() {
var err error
p.TimeTickChannelName, err = p.Load("msgChannel.chanNamePrefix.dataCoordTimeTick")
if err != nil {
panic(err)
}
}
func (p *ParamTable) initSegmentInfoChannelName() {
var err error
p.SegmentInfoChannelName, err = p.Load("msgChannel.chanNamePrefix.dataCoordSegmentInfo")
if err != nil {
panic(err)
}
}
func (p *ParamTable) initDataCoordSubscriptionName() {
var err error
p.DataCoordSubscriptionName, err = p.Load("msgChannel.subNamePrefix.dataCoordSubNamePrefix")
if err != nil {
panic(err)
}
}
func (p *ParamTable) initLogCfg() {
p.Log = log.Config{}
format, err := p.Load("log.format")
if err != nil {
panic(err)
}
p.Log.Format = format
level, err := p.Load("log.level")
if err != nil {
panic(err)
}
p.Log.Level = level
p.Log.File.MaxSize = p.ParseInt("log.file.maxSize")
p.Log.File.MaxBackups = p.ParseInt("log.file.maxBackups")
p.Log.File.MaxDays = p.ParseInt("log.file.maxAge")
rootPath, err := p.Load("log.file.rootPath")
if err != nil {
panic(err)
}
if len(rootPath) != 0 {
p.Log.File.Filename = path.Join(rootPath, "datacoord-"+strconv.FormatInt(p.NodeID, 10)+".log")
} else {
p.Log.File.Filename = ""
}
}
func (p *ParamTable) initFlushStreamPosSubPath() {
subPath, err := p.Load("etcd.flushStreamPosSubPath")
if err != nil {
panic(err)
}
p.FlushStreamPosSubPath = subPath
}
func (p *ParamTable) initStatsStreamPosSubPath() {
subPath, err := p.Load("etcd.statsStreamPosSubPath")
if err != nil {
panic(err)
}
p.StatsStreamPosSubPath = subPath
}