component_param.go 29.7 KB
Newer Older
1 2 3 4 5 6 7 8 9 10 11 12 13 14
// 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 paramtable

import (
15
	"math"
16 17
	"os"
	"path"
18
	"strconv"
19
	"strings"
20
	"sync"
X
Xiaofan 已提交
21
	"sync/atomic"
22
	"time"
23

24
	"go.uber.org/zap"
G
godchen 已提交
25 26

	"github.com/milvus-io/milvus/internal/log"
27 28 29
)

const (
30 31
	// DefaultRetentionDuration defines the default duration for retention which is 5 days in seconds.
	DefaultRetentionDuration = 3600 * 24 * 5
32 33

	// DefaultIndexSliceSize defines the default slice size of index file when serializing.
34
	DefaultIndexSliceSize = 16
35
	DefaultGracefulTime   = 5000 //ms
36 37
)

38
// ComponentParam is used to quickly and easily access all components' configurations.
39 40
type ComponentParam struct {
	ServiceParam
41
	once sync.Once
42

43
	CommonCfg commonConfig
44 45 46 47 48 49 50 51 52 53 54 55

	RootCoordCfg  rootCoordConfig
	ProxyCfg      proxyConfig
	QueryCoordCfg queryCoordConfig
	QueryNodeCfg  queryNodeConfig
	DataCoordCfg  dataCoordConfig
	DataNodeCfg   dataNodeConfig
	IndexCoordCfg indexCoordConfig
	IndexNodeCfg  indexNodeConfig
}

// InitOnce initialize once
56
func (p *ComponentParam) InitOnce() {
57 58 59 60 61
	p.once.Do(func() {
		p.Init()
	})
}

62
// Init initialize the global param table
63 64
func (p *ComponentParam) Init() {
	p.ServiceParam.Init()
65

66
	p.CommonCfg.init(&p.BaseTable)
67

68 69 70 71 72 73 74 75
	p.RootCoordCfg.init(&p.BaseTable)
	p.ProxyCfg.init(&p.BaseTable)
	p.QueryCoordCfg.init(&p.BaseTable)
	p.QueryNodeCfg.init(&p.BaseTable)
	p.DataCoordCfg.init(&p.BaseTable)
	p.DataNodeCfg.init(&p.BaseTable)
	p.IndexCoordCfg.init(&p.BaseTable)
	p.IndexNodeCfg.init(&p.BaseTable)
76 77
}

78
// SetLogConfig set log config with given role
79
func (p *ComponentParam) SetLogConfig(role string) {
80 81
	p.BaseTable.RoleName = role
	p.BaseTable.SetLogConfig()
82 83
}

84 85 86 87 88 89 90 91
func (p *ComponentParam) RocksmqEnable() bool {
	return p.RocksmqCfg.Path != ""
}

func (p *ComponentParam) PulsarEnable() bool {
	return p.PulsarCfg.Address != ""
}

J
jaime 已提交
92 93 94 95
func (p *ComponentParam) KafkaEnable() bool {
	return p.KafkaCfg.Address != ""
}

96
///////////////////////////////////////////////////////////////////////////////
97
// --- common ---
98
type commonConfig struct {
99
	Base *BaseTable
100

101
	ClusterPrefix string
102

103 104
	ProxySubName string

105 106 107 108 109
	RootCoordTimeTick   string
	RootCoordStatistics string
	RootCoordDml        string
	RootCoordDelta      string
	RootCoordSubName    string
110

111 112 113 114
	QueryCoordSearch       string
	QueryCoordSearchResult string
	QueryCoordTimeTick     string
	QueryNodeStats         string
115
	QueryNodeSubName       string
116

117 118 119 120
	DataCoordStatistic   string
	DataCoordTimeTick    string
	DataCoordSegmentInfo string
	DataCoordSubName     string
121
	DataNodeSubName      string
122 123 124 125

	DefaultPartitionName string
	DefaultIndexName     string
	RetentionDuration    int64
X
Xiaofan 已提交
126
	EntityExpirationTTL  time.Duration
127

128
	IndexSliceSize int64
129 130 131 132
	GracefulTime   int64

	StorageType string
	SimdType    string
C
codeman 已提交
133 134

	AuthorizationEnabled bool
135 136
}

137
func (p *commonConfig) init(base *BaseTable) {
138
	p.Base = base
139

140 141
	// must init cluster prefix first
	p.initClusterPrefix()
142 143
	p.initProxySubName()

144 145 146 147 148 149 150 151 152 153
	p.initRootCoordTimeTick()
	p.initRootCoordStatistics()
	p.initRootCoordDml()
	p.initRootCoordDelta()
	p.initRootCoordSubName()

	p.initQueryCoordSearch()
	p.initQueryCoordSearchResult()
	p.initQueryCoordTimeTick()
	p.initQueryNodeStats()
154
	p.initQueryNodeSubName()
155 156 157 158 159

	p.initDataCoordStatistic()
	p.initDataCoordTimeTick()
	p.initDataCoordSegmentInfo()
	p.initDataCoordSubName()
160
	p.initDataNodeSubName()
161 162 163 164

	p.initDefaultPartitionName()
	p.initDefaultIndexName()
	p.initRetentionDuration()
X
Xiaofan 已提交
165
	p.initEntityExpiration()
166 167

	p.initSimdType()
168
	p.initIndexSliceSize()
169
	p.initGracefulTime()
J
jaime 已提交
170
	p.initStorageType()
C
codeman 已提交
171 172

	p.initEnableAuthorization()
173 174
}

175 176 177 178 179 180
func (p *commonConfig) initClusterPrefix() {
	keys := []string{
		"common.chanNamePrefix.cluster",
		"msgChannel.chanNamePrefix.cluster",
	}
	str, err := p.Base.Load2(keys)
181 182 183
	if err != nil {
		panic(err)
	}
184
	p.ClusterPrefix = str
185 186
}

187 188
func (p *commonConfig) initChanNamePrefix(keys []string) string {
	value, err := p.Base.Load2(keys)
189 190 191
	if err != nil {
		panic(err)
	}
192 193
	s := []string{p.ClusterPrefix, value}
	return strings.Join(s, "-")
194 195
}

196
// --- proxy ---
197 198 199 200 201 202
func (p *commonConfig) initProxySubName() {
	keys := []string{
		"common.subNamePrefix.proxySubNamePrefix",
		"msgChannel.subNamePrefix.proxySubNamePrefix",
	}
	p.ProxySubName = p.initChanNamePrefix(keys)
203 204
}

205
// --- rootcoord ---
X
Xiaofan 已提交
206
// Deprecate
207 208 209 210 211 212
func (p *commonConfig) initRootCoordTimeTick() {
	keys := []string{
		"common.chanNamePrefix.rootCoordTimeTick",
		"msgChannel.chanNamePrefix.rootCoordTimeTick",
	}
	p.RootCoordTimeTick = p.initChanNamePrefix(keys)
213 214
}

215 216 217 218 219 220
func (p *commonConfig) initRootCoordStatistics() {
	keys := []string{
		"common.chanNamePrefix.rootCoordStatistics",
		"msgChannel.chanNamePrefix.rootCoordStatistics",
	}
	p.RootCoordStatistics = p.initChanNamePrefix(keys)
221 222
}

223 224 225 226 227 228
func (p *commonConfig) initRootCoordDml() {
	keys := []string{
		"common.chanNamePrefix.rootCoordDml",
		"msgChannel.chanNamePrefix.rootCoordDml",
	}
	p.RootCoordDml = p.initChanNamePrefix(keys)
229 230
}

231 232 233 234 235 236
func (p *commonConfig) initRootCoordDelta() {
	keys := []string{
		"common.chanNamePrefix.rootCoordDelta",
		"msgChannel.chanNamePrefix.rootCoordDelta",
	}
	p.RootCoordDelta = p.initChanNamePrefix(keys)
237 238
}

239 240 241 242 243 244
func (p *commonConfig) initRootCoordSubName() {
	keys := []string{
		"common.subNamePrefix.rootCoordSubNamePrefix",
		"msgChannel.subNamePrefix.rootCoordSubNamePrefix",
	}
	p.RootCoordSubName = p.initChanNamePrefix(keys)
245 246 247
}

// --- querycoord ---
248 249 250 251 252 253
func (p *commonConfig) initQueryCoordSearch() {
	keys := []string{
		"common.chanNamePrefix.search",
		"msgChannel.chanNamePrefix.search",
	}
	p.QueryCoordSearch = p.initChanNamePrefix(keys)
254 255
}

X
Xiaofan 已提交
256
// Deprecated, search result use grpc instead of a result channel
257 258 259 260 261 262
func (p *commonConfig) initQueryCoordSearchResult() {
	keys := []string{
		"common.chanNamePrefix.searchResult",
		"msgChannel.chanNamePrefix.searchResult",
	}
	p.QueryCoordSearchResult = p.initChanNamePrefix(keys)
263 264
}

X
Xiaofan 已提交
265
// Deprecate
266 267 268 269 270 271
func (p *commonConfig) initQueryCoordTimeTick() {
	keys := []string{
		"common.chanNamePrefix.queryTimeTick",
		"msgChannel.chanNamePrefix.queryTimeTick",
	}
	p.QueryCoordTimeTick = p.initChanNamePrefix(keys)
272 273 274
}

// --- querynode ---
275 276 277 278 279 280
func (p *commonConfig) initQueryNodeStats() {
	keys := []string{
		"common.chanNamePrefix.queryNodeStats",
		"msgChannel.chanNamePrefix.queryNodeStats",
	}
	p.QueryNodeStats = p.initChanNamePrefix(keys)
281 282
}

283 284 285 286 287 288
func (p *commonConfig) initQueryNodeSubName() {
	keys := []string{
		"common.subNamePrefix.queryNodeSubNamePrefix",
		"msgChannel.subNamePrefix.queryNodeSubNamePrefix",
	}
	p.QueryNodeSubName = p.initChanNamePrefix(keys)
289 290
}

291
// --- datacoord ---
292 293 294 295 296 297
func (p *commonConfig) initDataCoordStatistic() {
	keys := []string{
		"common.chanNamePrefix.dataCoordStatistic",
		"msgChannel.chanNamePrefix.dataCoordStatistic",
	}
	p.DataCoordStatistic = p.initChanNamePrefix(keys)
298 299
}

X
Xiaofan 已提交
300
// Deprecate
301 302 303 304 305 306
func (p *commonConfig) initDataCoordTimeTick() {
	keys := []string{
		"common.chanNamePrefix.dataCoordTimeTick",
		"msgChannel.chanNamePrefix.dataCoordTimeTick",
	}
	p.DataCoordTimeTick = p.initChanNamePrefix(keys)
307 308
}

309 310 311 312 313 314
func (p *commonConfig) initDataCoordSegmentInfo() {
	keys := []string{
		"common.chanNamePrefix.dataCoordSegmentInfo",
		"msgChannel.chanNamePrefix.dataCoordSegmentInfo",
	}
	p.DataCoordSegmentInfo = p.initChanNamePrefix(keys)
315 316
}

317 318 319 320 321 322
func (p *commonConfig) initDataCoordSubName() {
	keys := []string{
		"common.subNamePrefix.dataCoordSubNamePrefix",
		"msgChannel.subNamePrefix.dataCoordSubNamePrefix",
	}
	p.DataCoordSubName = p.initChanNamePrefix(keys)
323 324
}

325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344
func (p *commonConfig) initDataNodeSubName() {
	keys := []string{
		"common.subNamePrefix.dataNodeSubNamePrefix",
		"msgChannel.subNamePrefix.dataNodeSubNamePrefix",
	}
	p.DataNodeSubName = p.initChanNamePrefix(keys)
}

func (p *commonConfig) initDefaultPartitionName() {
	p.DefaultPartitionName = p.Base.LoadWithDefault("common.defaultPartitionName", "_default")
}

func (p *commonConfig) initDefaultIndexName() {
	p.DefaultIndexName = p.Base.LoadWithDefault("common.defaultIndexName", "_default_idx")
}

func (p *commonConfig) initRetentionDuration() {
	p.RetentionDuration = p.Base.ParseInt64WithDefault("common.retentionDuration", DefaultRetentionDuration)
}

X
Xiaofan 已提交
345 346 347 348 349 350 351 352 353 354 355 356 357 358 359
func (p *commonConfig) initEntityExpiration() {
	ttl := p.Base.ParseInt64WithDefault("common.entityExpiration", -1)
	if ttl < 0 {
		p.EntityExpirationTTL = -1
		return
	}

	// make sure ttl is larger than retention duration to ensure time travel works
	if ttl > p.RetentionDuration {
		p.EntityExpirationTTL = time.Duration(ttl) * time.Second
	} else {
		p.EntityExpirationTTL = time.Duration(p.RetentionDuration) * time.Second
	}
}

360 361 362 363 364 365
func (p *commonConfig) initSimdType() {
	keys := []string{
		"common.simdType",
		"knowhere.simdType",
	}
	p.SimdType = p.Base.LoadWithDefault2(keys, "auto")
366 367
}

368 369 370 371
func (p *commonConfig) initIndexSliceSize() {
	p.IndexSliceSize = p.Base.ParseInt64WithDefault("common.indexSliceSize", DefaultIndexSliceSize)
}

372 373 374 375
func (p *commonConfig) initGracefulTime() {
	p.GracefulTime = p.Base.ParseInt64WithDefault("common.gracefulTime", DefaultGracefulTime)
}

J
jaime 已提交
376
func (p *commonConfig) initStorageType() {
X
Xiaofan 已提交
377
	p.StorageType = p.Base.LoadWithDefault("common.storageType", "minio")
J
jaime 已提交
378 379
}

C
codeman 已提交
380 381 382 383
func (p *commonConfig) initEnableAuthorization() {
	p.AuthorizationEnabled = p.Base.ParseBool("common.security.authorizationEnabled", false)
}

384 385 386
///////////////////////////////////////////////////////////////////////////////
// --- rootcoord ---
type rootCoordConfig struct {
387
	Base *BaseTable
388 389 390 391

	Address string
	Port    int

392 393 394 395 396 397 398 399 400
	DmlChannelNum                   int64
	MaxPartitionNum                 int64
	MinSegmentSizeToEnableIndex     int64
	ImportTaskExpiration            float64
	ImportTaskRetention             float64
	ImportSegmentStateCheckInterval float64
	ImportSegmentStateWaitLimit     float64
	ImportIndexCheckInterval        float64
	ImportIndexWaitLimit            float64
401

402 403 404
	// --- ETCD Path ---
	ImportTaskSubPath string

405 406 407 408
	CreatedTime time.Time
	UpdatedTime time.Time
}

409 410 411 412 413
func (p *rootCoordConfig) init(base *BaseTable) {
	p.Base = base
	p.DmlChannelNum = p.Base.ParseInt64WithDefault("rootCoord.dmlChannelNum", 256)
	p.MaxPartitionNum = p.Base.ParseInt64WithDefault("rootCoord.maxPartitionNum", 4096)
	p.MinSegmentSizeToEnableIndex = p.Base.ParseInt64WithDefault("rootCoord.minSegmentSizeToEnableIndex", 1024)
414 415
	p.ImportTaskExpiration = p.Base.ParseFloatWithDefault("rootCoord.importTaskExpiration", 3600)
	p.ImportTaskRetention = p.Base.ParseFloatWithDefault("rootCoord.importTaskRetention", 3600*24)
416 417
	p.ImportSegmentStateCheckInterval = p.Base.ParseFloatWithDefault("rootCoord.importSegmentStateCheckInterval", 10)
	p.ImportSegmentStateWaitLimit = p.Base.ParseFloatWithDefault("rootCoord.importSegmentStateWaitLimit", 60)
418 419
	p.ImportIndexCheckInterval = p.Base.ParseFloatWithDefault("rootCoord.importIndexCheckInterval", 60*5)
	p.ImportIndexWaitLimit = p.Base.ParseFloatWithDefault("rootCoord.importIndexWaitLimit", 60*20)
420
	p.ImportTaskSubPath = "importtask"
421 422 423 424 425
}

///////////////////////////////////////////////////////////////////////////////
// --- proxy ---
type proxyConfig struct {
426
	Base *BaseTable
427 428

	// NetworkPort & IP are not used
429 430
	NetworkPort    int
	IP             string
431 432 433 434
	NetworkAddress string

	Alias string

X
Xiaofan 已提交
435
	NodeID                   atomic.Value
436 437 438
	TimeTickInterval         time.Duration
	MsgStreamTimeTickBufSize int64
	MaxNameLength            int64
439
	MaxUsernameLength        int64
440
	MinPasswordLength        int64
441
	MaxPasswordLength        int64
442 443 444
	MaxFieldNum              int64
	MaxShardNum              int32
	MaxDimension             int64
445
	GinLogging               bool
446

447
	// required from QueryCoord
448 449 450 451 452 453 454 455 456
	SearchResultChannelNames   []string
	RetrieveResultChannelNames []string

	MaxTaskNum int64

	CreatedTime time.Time
	UpdatedTime time.Time
}

457 458
func (p *proxyConfig) init(base *BaseTable) {
	p.Base = base
X
Xiaofan 已提交
459
	p.NodeID.Store(UniqueID(0))
460 461 462 463
	p.initTimeTickInterval()

	p.initMsgStreamTimeTickBufSize()
	p.initMaxNameLength()
464
	p.initMinPasswordLength()
465 466
	p.initMaxUsernameLength()
	p.initMaxPasswordLength()
467 468 469 470 471
	p.initMaxFieldNum()
	p.initMaxShardNum()
	p.initMaxDimension()

	p.initMaxTaskNum()
472
	p.initGinLogging()
473 474 475 476 477 478 479 480
}

// InitAlias initialize Alias member.
func (p *proxyConfig) InitAlias(alias string) {
	p.Alias = alias
}

func (p *proxyConfig) initTimeTickInterval() {
481
	interval := p.Base.ParseIntWithDefault("proxy.timeTickInterval", 200)
482 483 484 485
	p.TimeTickInterval = time.Duration(interval) * time.Millisecond
}

func (p *proxyConfig) initMsgStreamTimeTickBufSize() {
486
	p.MsgStreamTimeTickBufSize = p.Base.ParseInt64WithDefault("proxy.msgStream.timeTick.bufSize", 512)
487 488 489
}

func (p *proxyConfig) initMaxNameLength() {
490
	str := p.Base.LoadWithDefault("proxy.maxNameLength", "255")
491 492 493 494 495 496 497
	maxNameLength, err := strconv.ParseInt(str, 10, 64)
	if err != nil {
		panic(err)
	}
	p.MaxNameLength = maxNameLength
}

498 499 500 501 502 503 504 505 506
func (p *proxyConfig) initMaxUsernameLength() {
	str := p.Base.LoadWithDefault("proxy.maxUsernameLength", "32")
	maxUsernameLength, err := strconv.ParseInt(str, 10, 64)
	if err != nil {
		panic(err)
	}
	p.MaxUsernameLength = maxUsernameLength
}

507 508 509 510 511 512 513 514 515
func (p *proxyConfig) initMinPasswordLength() {
	str := p.Base.LoadWithDefault("proxy.minPasswordLength", "6")
	minPasswordLength, err := strconv.ParseInt(str, 10, 64)
	if err != nil {
		panic(err)
	}
	p.MinPasswordLength = minPasswordLength
}

516 517 518 519 520 521 522 523 524
func (p *proxyConfig) initMaxPasswordLength() {
	str := p.Base.LoadWithDefault("proxy.maxPasswordLength", "256")
	maxPasswordLength, err := strconv.ParseInt(str, 10, 64)
	if err != nil {
		panic(err)
	}
	p.MaxPasswordLength = maxPasswordLength
}

525
func (p *proxyConfig) initMaxShardNum() {
526
	str := p.Base.LoadWithDefault("proxy.maxShardNum", "256")
527 528 529 530 531 532 533 534
	maxShardNum, err := strconv.ParseInt(str, 10, 64)
	if err != nil {
		panic(err)
	}
	p.MaxShardNum = int32(maxShardNum)
}

func (p *proxyConfig) initMaxFieldNum() {
535
	str := p.Base.LoadWithDefault("proxy.maxFieldNum", "64")
536 537 538 539 540 541 542 543
	maxFieldNum, err := strconv.ParseInt(str, 10, 64)
	if err != nil {
		panic(err)
	}
	p.MaxFieldNum = maxFieldNum
}

func (p *proxyConfig) initMaxDimension() {
544
	str := p.Base.LoadWithDefault("proxy.maxDimension", "32768")
545 546 547 548 549 550 551 552
	maxDimension, err := strconv.ParseInt(str, 10, 64)
	if err != nil {
		panic(err)
	}
	p.MaxDimension = maxDimension
}

func (p *proxyConfig) initMaxTaskNum() {
553
	p.MaxTaskNum = p.Base.ParseInt64WithDefault("proxy.maxTaskNum", 1024)
554 555
}

556 557 558 559 560
func (p *proxyConfig) initGinLogging() {
	// Gin logging is on by default.
	p.GinLogging = p.Base.ParseBool("proxy.ginLogging", true)
}

X
Xiaofan 已提交
561 562 563 564 565 566 567 568 569 570 571 572
func (p *proxyConfig) SetNodeID(id UniqueID) {
	p.NodeID.Store(id)
}

func (p *proxyConfig) GetNodeID() UniqueID {
	val := p.NodeID.Load()
	if val != nil {
		return val.(UniqueID)
	}
	return 0
}

573 574 575
///////////////////////////////////////////////////////////////////////////////
// --- querycoord ---
type queryCoordConfig struct {
576
	Base *BaseTable
577

X
Xiaofan 已提交
578 579 580
	Address string
	Port    int
	NodeID  atomic.Value
581 582 583 584 585 586 587 588 589 590 591 592 593 594

	CreatedTime time.Time
	UpdatedTime time.Time

	//---- Handoff ---
	AutoHandoff bool

	//---- Balance ---
	AutoBalance                         bool
	OverloadedMemoryThresholdPercentage float64
	BalanceIntervalSeconds              int64
	MemoryUsageMaxDifferencePercentage  float64
}

595 596
func (p *queryCoordConfig) init(base *BaseTable) {
	p.Base = base
X
Xiaofan 已提交
597
	p.NodeID.Store(UniqueID(0))
598 599 600 601 602 603 604 605 606 607 608
	//---- Handoff ---
	p.initAutoHandoff()

	//---- Balance ---
	p.initAutoBalance()
	p.initOverloadedMemoryThresholdPercentage()
	p.initBalanceIntervalSeconds()
	p.initMemoryUsageMaxDifferencePercentage()
}

func (p *queryCoordConfig) initAutoHandoff() {
609
	handoff, err := p.Base.Load("queryCoord.autoHandoff")
610 611 612 613 614 615 616 617 618 619
	if err != nil {
		panic(err)
	}
	p.AutoHandoff, err = strconv.ParseBool(handoff)
	if err != nil {
		panic(err)
	}
}

func (p *queryCoordConfig) initAutoBalance() {
620
	balanceStr := p.Base.LoadWithDefault("queryCoord.autoBalance", "false")
621 622 623 624 625 626 627 628
	autoBalance, err := strconv.ParseBool(balanceStr)
	if err != nil {
		panic(err)
	}
	p.AutoBalance = autoBalance
}

func (p *queryCoordConfig) initOverloadedMemoryThresholdPercentage() {
629
	overloadedMemoryThresholdPercentage := p.Base.LoadWithDefault("queryCoord.overloadedMemoryThresholdPercentage", "90")
630 631 632 633 634 635 636 637
	thresholdPercentage, err := strconv.ParseInt(overloadedMemoryThresholdPercentage, 10, 64)
	if err != nil {
		panic(err)
	}
	p.OverloadedMemoryThresholdPercentage = float64(thresholdPercentage) / 100
}

func (p *queryCoordConfig) initBalanceIntervalSeconds() {
638
	balanceInterval := p.Base.LoadWithDefault("queryCoord.balanceIntervalSeconds", "60")
639 640 641 642 643 644 645 646
	interval, err := strconv.ParseInt(balanceInterval, 10, 64)
	if err != nil {
		panic(err)
	}
	p.BalanceIntervalSeconds = interval
}

func (p *queryCoordConfig) initMemoryUsageMaxDifferencePercentage() {
647
	maxDiff := p.Base.LoadWithDefault("queryCoord.memoryUsageMaxDifferencePercentage", "30")
648 649 650 651 652 653 654
	diffPercentage, err := strconv.ParseInt(maxDiff, 10, 64)
	if err != nil {
		panic(err)
	}
	p.MemoryUsageMaxDifferencePercentage = float64(diffPercentage) / 100
}

X
Xiaofan 已提交
655 656 657 658 659 660 661 662 663 664 665 666
func (p *queryCoordConfig) SetNodeID(id UniqueID) {
	p.NodeID.Store(id)
}

func (p *queryCoordConfig) GetNodeID() UniqueID {
	val := p.NodeID.Load()
	if val != nil {
		return val.(UniqueID)
	}
	return 0
}

667 668 669
///////////////////////////////////////////////////////////////////////////////
// --- querynode ---
type queryNodeConfig struct {
670
	Base *BaseTable
671 672 673 674

	Alias         string
	QueryNodeIP   string
	QueryNodePort int64
X
Xiaofan 已提交
675
	NodeID        atomic.Value
676 677 678 679 680 681 682 683 684
	// TODO: remove cacheSize
	CacheSize int64 // deprecated

	FlowGraphMaxQueueLength int32
	FlowGraphMaxParallelism int32

	// stats
	StatsPublishInterval int

685
	SliceIndex int
686 687

	// segcore
688 689 690
	ChunkRows        int64
	SmallIndexNlist  int64
	SmallIndexNProbe int64
691 692 693 694 695 696

	CreatedTime time.Time
	UpdatedTime time.Time

	// memory limit
	OverloadedMemoryThresholdPercentage float64
G
godchen 已提交
697 698

	// cache limit
G
godchen 已提交
699 700
	CacheEnabled     bool
	CacheMemoryLimit int64
701 702 703 704 705 706

	GroupEnabled         bool
	MaxReceiveChanSize   int32
	MaxUnsolvedQueueSize int32
	MaxGroupNQ           int64
	TopKMergeRatio       float64
707 708
}

709 710
func (p *queryNodeConfig) init(base *BaseTable) {
	p.Base = base
X
Xiaofan 已提交
711
	p.NodeID.Store(UniqueID(0))
712 713 714 715 716 717 718
	p.initCacheSize()

	p.initFlowGraphMaxQueueLength()
	p.initFlowGraphMaxParallelism()

	p.initStatsPublishInterval()

719
	p.initSmallIndexParams()
720 721

	p.initOverloadedMemoryThresholdPercentage()
G
godchen 已提交
722

G
godchen 已提交
723 724
	p.initCacheMemoryLimit()
	p.initCacheEnabled()
725 726 727 728 729 730

	p.initGroupEnabled()
	p.initMaxReceiveChanSize()
	p.initMaxUnsolvedQueueSize()
	p.initMaxGroupNQ()
	p.initTopKMergeRatio()
731 732 733 734 735 736 737 738 739 740 741 742 743 744 745 746
}

// InitAlias initializes an alias for the QueryNode role.
func (p *queryNodeConfig) InitAlias(alias string) {
	p.Alias = alias
}

func (p *queryNodeConfig) initCacheSize() {
	defer log.Debug("init cacheSize", zap.Any("cacheSize (GB)", p.CacheSize))

	const defaultCacheSize = 32 // GB
	p.CacheSize = defaultCacheSize

	var err error
	cacheSize := os.Getenv("CACHE_SIZE")
	if cacheSize == "" {
747
		cacheSize, err = p.Base.Load("queryNode.cacheSize")
748 749 750 751 752 753 754 755 756 757 758 759 760 761
		if err != nil {
			return
		}
	}
	value, err := strconv.ParseInt(cacheSize, 10, 64)
	if err != nil {
		return
	}
	p.CacheSize = value
}

// advanced params
// stats
func (p *queryNodeConfig) initStatsPublishInterval() {
762
	p.StatsPublishInterval = p.Base.ParseIntWithDefault("queryNode.stats.publishInterval", 1000)
763 764 765 766
}

// dataSync:
func (p *queryNodeConfig) initFlowGraphMaxQueueLength() {
767
	p.FlowGraphMaxQueueLength = p.Base.ParseInt32WithDefault("queryNode.dataSync.flowGraph.maxQueueLength", 1024)
768 769 770
}

func (p *queryNodeConfig) initFlowGraphMaxParallelism() {
771
	p.FlowGraphMaxParallelism = p.Base.ParseInt32WithDefault("queryNode.dataSync.flowGraph.maxParallelism", 1024)
772 773
}

774
func (p *queryNodeConfig) initSmallIndexParams() {
775
	p.ChunkRows = p.Base.ParseInt64WithDefault("queryNode.segcore.chunkRows", 32768)
776 777 778 779 780 781 782 783 784 785 786 787 788 789 790 791 792 793 794 795 796 797 798 799 800 801
	if p.ChunkRows < 1024 {
		log.Warn("chunk rows can not be less than 1024, force set to 1024", zap.Any("current", p.ChunkRows))
		p.ChunkRows = 1024
	}

	// default NList is the first nlist
	var defaultNList int64
	for i := int64(0); i < p.ChunkRows; i++ {
		if math.Pow(2.0, float64(i)) > math.Sqrt(float64(p.ChunkRows)) {
			defaultNList = int64(math.Pow(2, float64(i)))
			break
		}
	}

	p.SmallIndexNlist = p.Base.ParseInt64WithDefault("queryNode.segcore.smallIndex.nlist", defaultNList)
	if p.SmallIndexNlist > p.ChunkRows/8 {
		log.Warn("small index nlist must smaller than chunkRows/8, force set to", zap.Any("nliit", p.ChunkRows/8))
		p.SmallIndexNlist = p.ChunkRows / 8
	}

	defaultNprobe := p.SmallIndexNlist / 16
	p.SmallIndexNProbe = p.Base.ParseInt64WithDefault("queryNode.segcore.smallIndex.nprobe", defaultNprobe)
	if p.SmallIndexNProbe > p.SmallIndexNlist {
		log.Warn("small index nprobe must smaller than nlist, force set to", zap.Any("nprobe", p.SmallIndexNlist))
		p.SmallIndexNProbe = p.SmallIndexNlist
	}
802 803 804
}

func (p *queryNodeConfig) initOverloadedMemoryThresholdPercentage() {
805
	overloadedMemoryThresholdPercentage := p.Base.LoadWithDefault("queryCoord.overloadedMemoryThresholdPercentage", "90")
806 807 808 809 810 811 812
	thresholdPercentage, err := strconv.ParseInt(overloadedMemoryThresholdPercentage, 10, 64)
	if err != nil {
		panic(err)
	}
	p.OverloadedMemoryThresholdPercentage = float64(thresholdPercentage) / 100
}

G
godchen 已提交
813 814 815 816 817 818 819 820
func (p *queryNodeConfig) initCacheMemoryLimit() {
	overloadedMemoryThresholdPercentage := p.Base.LoadWithDefault("queryNode.cache.memoryLimit", "2147483648")
	cacheMemoryLimit, err := strconv.ParseInt(overloadedMemoryThresholdPercentage, 10, 64)
	if err != nil {
		panic(err)
	}
	p.CacheMemoryLimit = cacheMemoryLimit
}
X
Xiaofan 已提交
821

G
godchen 已提交
822 823 824 825
func (p *queryNodeConfig) initCacheEnabled() {
	var err error
	cacheEnabled := p.Base.LoadWithDefault("queryNode.cache.enabled", "true")
	p.CacheEnabled, err = strconv.ParseBool(cacheEnabled)
G
godchen 已提交
826 827 828 829 830
	if err != nil {
		panic(err)
	}
}

831 832 833 834 835 836 837 838 839 840 841 842 843 844 845 846 847 848 849 850
func (p *queryNodeConfig) initGroupEnabled() {
	p.GroupEnabled = p.Base.ParseBool("queryNode.grouping.enabled", true)
}

func (p *queryNodeConfig) initMaxReceiveChanSize() {
	p.MaxReceiveChanSize = p.Base.ParseInt32WithDefault("queryNode.grouping.receiveChanSize", 10240)
}

func (p *queryNodeConfig) initMaxUnsolvedQueueSize() {
	p.MaxUnsolvedQueueSize = p.Base.ParseInt32WithDefault("queryNode.grouping.unsolvedQueueSize", 10240)
}

func (p *queryNodeConfig) initMaxGroupNQ() {
	p.MaxGroupNQ = p.Base.ParseInt64WithDefault("queryNode.grouping.maxNQ", 1000)
}

func (p *queryNodeConfig) initTopKMergeRatio() {
	p.TopKMergeRatio = p.Base.ParseFloatWithDefault("queryNode.grouping.topKMergeRatio", 10.0)
}

X
Xiaofan 已提交
851 852 853 854 855 856 857 858 859 860 861 862
func (p *queryNodeConfig) SetNodeID(id UniqueID) {
	p.NodeID.Store(id)
}

func (p *queryNodeConfig) GetNodeID() UniqueID {
	val := p.NodeID.Load()
	if val != nil {
		return val.(UniqueID)
	}
	return 0
}

863 864 865
///////////////////////////////////////////////////////////////////////////////
// --- datacoord ---
type dataCoordConfig struct {
866
	Base *BaseTable
867

X
Xiaofan 已提交
868
	NodeID atomic.Value
869 870 871 872 873 874

	IP      string
	Port    int
	Address string

	// --- ETCD ---
X
XuanYang-cn 已提交
875
	ChannelWatchSubPath string
876 877 878 879 880

	// --- SEGMENTS ---
	SegmentMaxSize          float64
	SegmentSealProportion   float64
	SegAssignmentExpiration int64
881
	SegmentMaxLifetime      time.Duration
882 883 884 885 886

	CreatedTime time.Time
	UpdatedTime time.Time

	EnableCompaction        bool
887
	EnableAutoCompaction    atomic.Value
888 889 890 891 892 893 894 895
	EnableGarbageCollection bool

	// Garbage Collection
	GCInterval         time.Duration
	GCMissingTolerance time.Duration
	GCDropTolerance    time.Duration
}

896 897
func (p *dataCoordConfig) init(base *BaseTable) {
	p.Base = base
898 899 900 901 902
	p.initChannelWatchPrefix()

	p.initSegmentMaxSize()
	p.initSegmentSealProportion()
	p.initSegAssignmentExpiration()
903
	p.initSegmentMaxLifetime()
904 905 906 907 908 909 910 911 912 913 914

	p.initEnableCompaction()
	p.initEnableAutoCompaction()

	p.initEnableGarbageCollection()
	p.initGCInterval()
	p.initGCMissingTolerance()
	p.initGCDropTolerance()
}

func (p *dataCoordConfig) initSegmentMaxSize() {
915
	p.SegmentMaxSize = p.Base.ParseFloatWithDefault("dataCoord.segment.maxSize", 512.0)
916 917 918
}

func (p *dataCoordConfig) initSegmentSealProportion() {
919
	p.SegmentSealProportion = p.Base.ParseFloatWithDefault("dataCoord.segment.sealProportion", 0.25)
920 921 922
}

func (p *dataCoordConfig) initSegAssignmentExpiration() {
923
	p.SegAssignmentExpiration = p.Base.ParseInt64WithDefault("dataCoord.segment.assignmentExpiration", 2000)
924 925
}

926 927 928 929
func (p *dataCoordConfig) initSegmentMaxLifetime() {
	p.SegmentMaxLifetime = time.Duration(p.Base.ParseInt64WithDefault("dataCoord.segment.maxLife", 24*60*60)) * time.Second
}

930 931 932 933 934 935 936
func (p *dataCoordConfig) initChannelWatchPrefix() {
	// WARN: this value should not be put to milvus.yaml. It's a default value for channel watch path.
	// This will be removed after we reconstruct our config module.
	p.ChannelWatchSubPath = "channelwatch"
}

func (p *dataCoordConfig) initEnableCompaction() {
937
	p.EnableCompaction = p.Base.ParseBool("dataCoord.enableCompaction", false)
938 939 940 941
}

// -- GC --
func (p *dataCoordConfig) initEnableGarbageCollection() {
942
	p.EnableGarbageCollection = p.Base.ParseBool("dataCoord.enableGarbageCollection", false)
943 944 945
}

func (p *dataCoordConfig) initGCInterval() {
946
	p.GCInterval = time.Duration(p.Base.ParseInt64WithDefault("dataCoord.gc.interval", 60*60)) * time.Second
947 948 949
}

func (p *dataCoordConfig) initGCMissingTolerance() {
950
	p.GCMissingTolerance = time.Duration(p.Base.ParseInt64WithDefault("dataCoord.gc.missingTolerance", 24*60*60)) * time.Second
951 952 953
}

func (p *dataCoordConfig) initGCDropTolerance() {
954
	p.GCDropTolerance = time.Duration(p.Base.ParseInt64WithDefault("dataCoord.gc.dropTolerance", 24*60*60)) * time.Second
955 956 957
}

func (p *dataCoordConfig) initEnableAutoCompaction() {
958 959 960 961 962 963 964 965 966 967 968 969 970
	p.EnableAutoCompaction.Store(p.Base.ParseBool("dataCoord.compaction.enableAutoCompaction", false))
}

func (p *dataCoordConfig) SetEnableAutoCompaction(enable bool) {
	p.EnableAutoCompaction.Store(enable)
}

func (p *dataCoordConfig) GetEnableAutoCompaction() bool {
	enable := p.EnableAutoCompaction.Load()
	if enable != nil {
		return enable.(bool)
	}
	return false
971 972
}

X
Xiaofan 已提交
973 974 975 976 977 978 979 980 981 982 983 984
func (p *dataCoordConfig) SetNodeID(id UniqueID) {
	p.NodeID.Store(id)
}

func (p *dataCoordConfig) GetNodeID() UniqueID {
	val := p.NodeID.Load()
	if val != nil {
		return val.(UniqueID)
	}
	return 0
}

985 986 987
///////////////////////////////////////////////////////////////////////////////
// --- datanode ---
type dataNodeConfig struct {
988
	Base *BaseTable
989

X
Xiaofan 已提交
990 991 992
	// ID of the current node
	//NodeID atomic.Value
	NodeID atomic.Value
993 994 995 996 997 998 999 1000 1001 1002 1003 1004 1005 1006 1007 1008 1009 1010 1011 1012
	// IP of the current DataNode
	IP string

	// Port of the current DataNode
	Port                    int
	FlowGraphMaxQueueLength int32
	FlowGraphMaxParallelism int32
	FlushInsertBufferSize   int64
	InsertBinlogRootPath    string
	StatsBinlogRootPath     string
	DeleteBinlogRootPath    string
	Alias                   string // Different datanode in one machine

	// etcd
	ChannelWatchSubPath string

	CreatedTime time.Time
	UpdatedTime time.Time
}

1013 1014
func (p *dataNodeConfig) init(base *BaseTable) {
	p.Base = base
X
Xiaofan 已提交
1015
	p.NodeID.Store(UniqueID(0))
1016 1017 1018 1019 1020 1021 1022 1023 1024 1025 1026 1027 1028 1029 1030 1031
	p.initFlowGraphMaxQueueLength()
	p.initFlowGraphMaxParallelism()
	p.initFlushInsertBufferSize()
	p.initInsertBinlogRootPath()
	p.initStatsBinlogRootPath()
	p.initDeleteBinlogRootPath()

	p.initChannelWatchPath()
}

// InitAlias init this DataNode alias
func (p *dataNodeConfig) InitAlias(alias string) {
	p.Alias = alias
}

func (p *dataNodeConfig) initFlowGraphMaxQueueLength() {
1032
	p.FlowGraphMaxQueueLength = p.Base.ParseInt32WithDefault("dataNode.dataSync.flowGraph.maxQueueLength", 1024)
1033 1034 1035
}

func (p *dataNodeConfig) initFlowGraphMaxParallelism() {
1036
	p.FlowGraphMaxParallelism = p.Base.ParseInt32WithDefault("dataNode.dataSync.flowGraph.maxParallelism", 1024)
1037 1038 1039
}

func (p *dataNodeConfig) initFlushInsertBufferSize() {
1040
	p.FlushInsertBufferSize = p.Base.ParseInt64("_DATANODE_INSERTBUFSIZE")
1041 1042 1043
}

func (p *dataNodeConfig) initInsertBinlogRootPath() {
1044
	// GOOSE TODO: rootPath change to TenentID
1045
	rootPath, err := p.Base.Load("minio.rootPath")
1046 1047 1048 1049 1050 1051 1052
	if err != nil {
		panic(err)
	}
	p.InsertBinlogRootPath = path.Join(rootPath, "insert_log")
}

func (p *dataNodeConfig) initStatsBinlogRootPath() {
1053
	rootPath, err := p.Base.Load("minio.rootPath")
1054 1055 1056 1057 1058 1059 1060
	if err != nil {
		panic(err)
	}
	p.StatsBinlogRootPath = path.Join(rootPath, "stats_log")
}

func (p *dataNodeConfig) initDeleteBinlogRootPath() {
1061
	rootPath, err := p.Base.Load("minio.rootPath")
1062 1063 1064 1065 1066 1067 1068 1069 1070 1071
	if err != nil {
		panic(err)
	}
	p.DeleteBinlogRootPath = path.Join(rootPath, "delta_log")
}

func (p *dataNodeConfig) initChannelWatchPath() {
	p.ChannelWatchSubPath = "channelwatch"
}

X
Xiaofan 已提交
1072 1073 1074 1075 1076 1077 1078 1079 1080 1081 1082 1083
func (p *dataNodeConfig) SetNodeID(id UniqueID) {
	p.NodeID.Store(id)
}

func (p *dataNodeConfig) GetNodeID() UniqueID {
	val := p.NodeID.Load()
	if val != nil {
		return val.(UniqueID)
	}
	return 0
}

1084 1085 1086
///////////////////////////////////////////////////////////////////////////////
// --- indexcoord ---
type indexCoordConfig struct {
1087
	Base *BaseTable
1088 1089 1090 1091 1092 1093 1094 1095 1096 1097

	Address string
	Port    int

	IndexStorageRootPath string

	CreatedTime time.Time
	UpdatedTime time.Time
}

1098 1099
func (p *indexCoordConfig) init(base *BaseTable) {
	p.Base = base
1100 1101 1102 1103 1104 1105

	p.initIndexStorageRootPath()
}

// initIndexStorageRootPath initializes the root path of index files.
func (p *indexCoordConfig) initIndexStorageRootPath() {
1106
	rootPath, err := p.Base.Load("minio.rootPath")
1107 1108 1109 1110 1111 1112 1113 1114 1115
	if err != nil {
		panic(err)
	}
	p.IndexStorageRootPath = path.Join(rootPath, "index_files")
}

///////////////////////////////////////////////////////////////////////////////
// --- indexnode ---
type indexNodeConfig struct {
1116
	Base *BaseTable
1117 1118 1119 1120 1121

	IP      string
	Address string
	Port    int

X
Xiaofan 已提交
1122 1123 1124
	NodeID atomic.Value

	Alias string
1125 1126 1127 1128 1129 1130 1131

	IndexStorageRootPath string

	CreatedTime time.Time
	UpdatedTime time.Time
}

1132 1133
func (p *indexNodeConfig) init(base *BaseTable) {
	p.Base = base
X
Xiaofan 已提交
1134
	p.NodeID.Store(UniqueID(0))
1135 1136 1137 1138 1139 1140 1141 1142 1143
	p.initIndexStorageRootPath()
}

// InitAlias initializes an alias for the IndexNode role.
func (p *indexNodeConfig) InitAlias(alias string) {
	p.Alias = alias
}

func (p *indexNodeConfig) initIndexStorageRootPath() {
1144
	rootPath, err := p.Base.Load("minio.rootPath")
1145 1146 1147 1148 1149
	if err != nil {
		panic(err)
	}
	p.IndexStorageRootPath = path.Join(rootPath, "index_files")
}
X
Xiaofan 已提交
1150 1151 1152 1153 1154 1155 1156 1157 1158 1159 1160 1161

func (p *indexNodeConfig) SetNodeID(id UniqueID) {
	p.NodeID.Store(id)
}

func (p *indexNodeConfig) GetNodeID() UniqueID {
	val := p.NodeID.Load()
	if val != nil {
		return val.(UniqueID)
	}
	return 0
}