milvus/internal/master/param_table.go

package master

import (
	"log"
	"strconv"
	"strings"

	"github.com/zilliztech/milvus-distributed/internal/util/paramtable"
	"github.com/zilliztech/milvus-distributed/internal/util/tsoutil"
	"github.com/zilliztech/milvus-distributed/internal/util/typeutil"
)

type ParamTable struct {
	paramtable.BaseTable

	Address string
	Port    int

	EtcdAddress   string
	EtcdRootPath  string
	PulsarAddress string

	// nodeID
	ProxyIDList     []typeutil.UniqueID
	WriteNodeIDList []typeutil.UniqueID

	TopicNum                    int
	QueryNodeNum                int
	SoftTimeTickBarrierInterval typeutil.Timestamp

	// segment
	SegmentSize           float64
	SegmentSizeFactor     float64
	DefaultRecordSize     int64
	MinSegIDAssignCnt     int64
	MaxSegIDAssignCnt     int64
	SegIDAssignExpiration int64

	// msgChannel
	ProxyTimeTickChannelNames     []string
	WriteNodeTimeTickChannelNames []string
	DDChannelNames                []string
	InsertChannelNames            []string
	K2SChannelNames               []string
	QueryNodeStatsChannelName     string
	MsgChannelSubName             string

	MaxPartitionNum     int64
	DefaultPartitionTag string
}

var Params ParamTable

func (p *ParamTable) Init() {
	// load yaml
	p.BaseTable.Init()
	err := p.LoadYaml("milvus.yaml")
	if err != nil {
		panic(err)
	}
	err = p.LoadYaml("advanced/channel.yaml")
	if err != nil {
		panic(err)
	}
	err = p.LoadYaml("advanced/master.yaml")
	if err != nil {
		panic(err)
	}
	err = p.LoadYaml("advanced/common.yaml")
	if err != nil {
		panic(err)
	}

	// set members
	p.initAddress()
	p.initPort()

	p.initEtcdAddress()
	p.initEtcdRootPath()
	p.initPulsarAddress()

	p.initProxyIDList()
	p.initWriteNodeIDList()

	p.initTopicNum()
	p.initQueryNodeNum()
	p.initSoftTimeTickBarrierInterval()

	p.initSegmentSize()
	p.initSegmentSizeFactor()
	p.initDefaultRecordSize()
	p.initMinSegIDAssignCnt()
	p.initMaxSegIDAssignCnt()
	p.initSegIDAssignExpiration()

	p.initProxyTimeTickChannelNames()
	p.initWriteNodeTimeTickChannelNames()
	p.initInsertChannelNames()
	p.initDDChannelNames()
	p.initK2SChannelNames()
	p.initQueryNodeStatsChannelName()
	p.initMsgChannelSubName()
	p.initMaxPartitionNum()
	p.initDefaultPartitionTag()
}

func (p *ParamTable) initAddress() {
	masterAddress, err := p.Load("master.address")
	if err != nil {
		panic(err)
	}
	p.Address = masterAddress
}

func (p *ParamTable) initPort() {
	masterPort, err := p.Load("master.port")
	if err != nil {
		panic(err)
	}
	port, err := strconv.Atoi(masterPort)
	if err != nil {
		panic(err)
	}
	p.Port = port
}

func (p *ParamTable) initEtcdAddress() {
	addr, err := p.Load("_EtcdAddress")
	if err != nil {
		panic(err)
	}
	p.EtcdAddress = addr
}

func (p *ParamTable) initPulsarAddress() {
	addr, err := p.Load("_PulsarAddress")
	if err != nil {
		panic(err)
	}
	p.PulsarAddress = addr
}

func (p *ParamTable) initEtcdRootPath() {
	path, err := p.Load("etcd.rootpath")
	if err != nil {
		panic(err)
	}
	p.EtcdRootPath = path
}

func (p *ParamTable) initTopicNum() {
	insertChannelRange, err := p.Load("msgChannel.channelRange.insert")
	if err != nil {
		panic(err)
	}

	channelRange := strings.Split(insertChannelRange, ",")
	if len(channelRange) != 2 {
		panic("Illegal channel range num")
	}
	channelBegin, err := strconv.Atoi(channelRange[0])
	if err != nil {
		panic(err)
	}
	channelEnd, err := strconv.Atoi(channelRange[1])
	if err != nil {
		panic(err)
	}
	if channelBegin < 0 || channelEnd < 0 {
		panic("Illegal channel range value")
	}
	if channelBegin > channelEnd {
		panic("Illegal channel range value")
	}
	p.TopicNum = channelEnd
}

func (p *ParamTable) initSegmentSize() {
	threshold, err := p.Load("master.segment.size")
	if err != nil {
		panic(err)
	}
	segmentThreshold, err := strconv.ParseFloat(threshold, 64)
	if err != nil {
		panic(err)
	}
	p.SegmentSize = segmentThreshold
}

func (p *ParamTable) initSegmentSizeFactor() {
	segFactor, err := p.Load("master.segment.sizeFactor")
	if err != nil {
		panic(err)
	}
	factor, err := strconv.ParseFloat(segFactor, 64)
	if err != nil {
		panic(err)
	}
	p.SegmentSizeFactor = factor
}

func (p *ParamTable) initDefaultRecordSize() {
	size, err := p.Load("master.segment.defaultSizePerRecord")
	if err != nil {
		panic(err)
	}
	res, err := strconv.ParseInt(size, 10, 64)
	if err != nil {
		panic(err)
	}
	p.DefaultRecordSize = res
}

func (p *ParamTable) initMinSegIDAssignCnt() {
	size, err := p.Load("master.segment.minIDAssignCnt")
	if err != nil {
		panic(err)
	}
	res, err := strconv.ParseInt(size, 10, 64)
	if err != nil {
		panic(err)
	}
	p.MinSegIDAssignCnt = res
}

func (p *ParamTable) initMaxSegIDAssignCnt() {
	size, err := p.Load("master.segment.maxIDAssignCnt")
	if err != nil {
		panic(err)
	}
	res, err := strconv.ParseInt(size, 10, 64)
	if err != nil {
		panic(err)
	}
	p.MaxSegIDAssignCnt = res
}

func (p *ParamTable) initSegIDAssignExpiration() {
	duration, err := p.Load("master.segment.IDAssignExpiration")
	if err != nil {
		panic(err)
	}
	res, err := strconv.ParseInt(duration, 10, 64)
	if err != nil {
		panic(err)
	}
	p.SegIDAssignExpiration = res
}

func (p *ParamTable) initQueryNodeNum() {
	id, err := p.Load("nodeID.queryNodeIDList")
	if err != nil {
		panic(err)
	}
	ids := strings.Split(id, ",")
	for _, i := range ids {
		_, err := strconv.ParseInt(i, 10, 64)
		if err != nil {
			log.Panicf("load proxy id list error, %s", err.Error())
		}
	}
	p.QueryNodeNum = len(ids)
}

func (p *ParamTable) initQueryNodeStatsChannelName() {
	channels, err := p.Load("msgChannel.chanNamePrefix.queryNodeStats")
	if err != nil {
		panic(err)
	}
	p.QueryNodeStatsChannelName = channels
}

func (p *ParamTable) initProxyIDList() {
	id, err := p.Load("nodeID.proxyIDList")
	if err != nil {
		log.Panicf("load proxy id list error, %s", err.Error())
	}
	ids := strings.Split(id, ",")
	idList := make([]typeutil.UniqueID, 0, len(ids))
	for _, i := range ids {
		v, err := strconv.ParseInt(i, 10, 64)
		if err != nil {
			log.Panicf("load proxy id list error, %s", err.Error())
		}
		idList = append(idList, typeutil.UniqueID(v))
	}
	p.ProxyIDList = idList
}

func (p *ParamTable) initProxyTimeTickChannelNames() {
	ch, err := p.Load("msgChannel.chanNamePrefix.proxyTimeTick")
	if err != nil {
		log.Panic(err)
	}
	id, err := p.Load("nodeID.proxyIDList")
	if err != nil {
		log.Panicf("load proxy id list error, %s", err.Error())
	}
	ids := strings.Split(id, ",")
	channels := make([]string, 0, len(ids))
	for _, i := range ids {
		_, err := strconv.ParseInt(i, 10, 64)
		if err != nil {
			log.Panicf("load proxy id list error, %s", err.Error())
		}
		channels = append(channels, ch+"-"+i)
	}
	p.ProxyTimeTickChannelNames = channels
}

func (p *ParamTable) initMsgChannelSubName() {
	name, err := p.Load("msgChannel.subNamePrefix.masterSubNamePrefix")
	if err != nil {
		log.Panic(err)
	}
	p.MsgChannelSubName = name
}

func (p *ParamTable) initSoftTimeTickBarrierInterval() {
	t, err := p.Load("master.timeSync.softTimeTickBarrierInterval")
	if err != nil {
		log.Panic(err)
	}
	v, err := strconv.ParseInt(t, 10, 64)
	if err != nil {
		log.Panic(err)
	}
	p.SoftTimeTickBarrierInterval = tsoutil.ComposeTS(v, 0)
}

func (p *ParamTable) initWriteNodeIDList() {
	id, err := p.Load("nodeID.writeNodeIDList")
	if err != nil {
		log.Panic(err)
	}
	ids := strings.Split(id, ",")
	idlist := make([]typeutil.UniqueID, 0, len(ids))
	for _, i := range ids {
		v, err := strconv.ParseInt(i, 10, 64)
		if err != nil {
			log.Panicf("load proxy id list error, %s", err.Error())
		}
		idlist = append(idlist, typeutil.UniqueID(v))
	}
	p.WriteNodeIDList = idlist
}

func (p *ParamTable) initWriteNodeTimeTickChannelNames() {
	ch, err := p.Load("msgChannel.chanNamePrefix.writeNodeTimeTick")
	if err != nil {
		log.Fatal(err)
	}
	id, err := p.Load("nodeID.writeNodeIDList")
	if err != nil {
		log.Panicf("load write node id list error, %s", err.Error())
	}
	ids := strings.Split(id, ",")
	channels := make([]string, 0, len(ids))
	for _, i := range ids {
		_, err := strconv.ParseInt(i, 10, 64)
		if err != nil {
			log.Panicf("load write node id list error, %s", err.Error())
		}
		channels = append(channels, ch+"-"+i)
	}
	p.WriteNodeTimeTickChannelNames = channels
}

func (p *ParamTable) initDDChannelNames() {
	ch, err := p.Load("msgChannel.chanNamePrefix.dataDefinition")
	if err != nil {
		log.Fatal(err)
	}
	id, err := p.Load("nodeID.queryNodeIDList")
	if err != nil {
		log.Panicf("load query node id list error, %s", err.Error())
	}
	ids := strings.Split(id, ",")
	channels := make([]string, 0, len(ids))
	for _, i := range ids {
		_, err := strconv.ParseInt(i, 10, 64)
		if err != nil {
			log.Panicf("load query node id list error, %s", err.Error())
		}
		channels = append(channels, ch+"-"+i)
	}
	p.DDChannelNames = channels
}

func (p *ParamTable) initInsertChannelNames() {
	ch, err := p.Load("msgChannel.chanNamePrefix.insert")
	if err != nil {
		log.Fatal(err)
	}
	channelRange, err := p.Load("msgChannel.channelRange.insert")
	if err != nil {
		panic(err)
	}

	chanRange := strings.Split(channelRange, ",")
	if len(chanRange) != 2 {
		panic("Illegal channel range num")
	}
	channelBegin, err := strconv.Atoi(chanRange[0])
	if err != nil {
		panic(err)
	}
	channelEnd, err := strconv.Atoi(chanRange[1])
	if err != nil {
		panic(err)
	}
	if channelBegin < 0 || channelEnd < 0 {
		panic("Illegal channel range value")
	}
	if channelBegin > channelEnd {
		panic("Illegal channel range value")
	}

	channels := make([]string, channelEnd-channelBegin)
	for i := 0; i < channelEnd-channelBegin; i++ {
		channels[i] = ch + "-" + strconv.Itoa(channelBegin+i)
	}
	p.InsertChannelNames = channels
}

func (p *ParamTable) initK2SChannelNames() {
	ch, err := p.Load("msgChannel.chanNamePrefix.k2s")
	if err != nil {
		log.Fatal(err)
	}
	id, err := p.Load("nodeID.writeNodeIDList")
	if err != nil {
		log.Panicf("load write node id list error, %s", err.Error())
	}
	ids := strings.Split(id, ",")
	channels := make([]string, 0, len(ids))
	for _, i := range ids {
		_, err := strconv.ParseInt(i, 10, 64)
		if err != nil {
			log.Panicf("load write node id list error, %s", err.Error())
		}
		channels = append(channels, ch+"-"+i)
	}
	p.K2SChannelNames = channels
}

func (p *ParamTable) initMaxPartitionNum() {
	str, err := p.Load("master.maxPartitionNum")
	if err != nil {
		panic(err)
	}
	maxPartitionNum, err := strconv.ParseInt(str, 10, 64)
	if err != nil {
		panic(err)
	}
	p.MaxPartitionNum = maxPartitionNum
}

func (p *ParamTable) initDefaultPartitionTag() {
	defaultTag, err := p.Load("common.defaultPartitionTag")
	if err != nil {
		panic(err)
	}

	p.DefaultPartitionTag = defaultTag
}