test/milvus
1
0
Fork 0
mirror of https://gitee.com/milvus-io/milvus.git synced 2024-11-30 10:59:32 +08:00
Code Issues Actions 11 Packages Projects Releases Wiki Activity

Update healthz to support standby state and fix rootcoord active-standby (#21078)

Browse Source 
Signed-off-by: wayblink <anyang.wang@zilliz.com>
This commit is contained in:
wayblink 2022-12-27 12:47:34 +08:00 committed by GitHub
parent cded2ce77c
commit 28b44b781f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 18 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
internal/management/healthz/healthz_handler.go
View File

@ -75,7 +75,7 @@ func (handler *HealthHandler) ServeHTTP(w http.ResponseWriter, r *http.Request)
Name: in.GetName(),
Code: code,
})
if code != commonpb.StateCode_Healthy {
if code != commonpb.StateCode_Healthy && code != commonpb.StateCode_StandBy {
resp.State = fmt.Sprintf("component %s state is %s", in.GetName(), code.String())
}
}

28
internal/rootcoord/root_coord.go
View File

@ -59,6 +59,7 @@ import (
"github.com/milvus-io/milvus/internal/util/dependency"
"github.com/milvus-io/milvus/internal/util/errorutil"
"github.com/milvus-io/milvus/internal/util/funcutil"
"github.com/milvus-io/milvus/internal/util/logutil"
"github.com/milvus-io/milvus/internal/util/metricsinfo"
"github.com/milvus-io/milvus/internal/util/paramtable"
"github.com/milvus-io/milvus/internal/util/retry"
@ -301,9 +302,6 @@ func (c *Core) Register() error {
c.session.Register()
if c.enableActiveStandBy {
c.session.ProcessActiveStandBy(c.activateFunc)
} else {
c.UpdateStateCode(commonpb.StateCode_Healthy)
log.Debug("RootCoord start successfully ", zap.String("State Code", commonpb.StateCode_Healthy.String()))
}
log.Info("RootCoord Register Finished")
go c.session.LivenessCheck(c.ctx, func() {
@ -626,14 +624,6 @@ func (c *Core) startInternal() error {
panic(err)
}
c.wg.Add(6)
go c.startTimeTickLoop()
go c.tsLoop()
go c.chanTimeTick.startWatch(&c.wg)
go c.importManager.cleanupLoop(&c.wg)
go c.importManager.sendOutTasksLoop(&c.wg)
go c.importManager.flipTaskStateLoop(&c.wg)
if Params.QuotaConfig.QuotaAndLimitsEnabled.GetAsBool() {
go c.quotaCenter.run()
}
@ -645,13 +635,29 @@ func (c *Core) startInternal() error {
c.activateFunc = func() {
// todo to complete
log.Info("rootcoord switch from standby to active, activating")
c.startServerLoop()
c.UpdateStateCode(commonpb.StateCode_Healthy)
}
c.UpdateStateCode(commonpb.StateCode_StandBy)
logutil.Logger(c.ctx).Info("rootcoord enter standby mode successfully")
} else {
c.startServerLoop()
c.UpdateStateCode(commonpb.StateCode_Healthy)
logutil.Logger(c.ctx).Info("rootcoord startup successfully")
}
return nil
}
func (c *Core) startServerLoop() {
c.wg.Add(6)
go c.startTimeTickLoop()
go c.tsLoop()
go c.chanTimeTick.startWatch(&c.wg)
go c.importManager.cleanupLoop(&c.wg)
go c.importManager.sendOutTasksLoop(&c.wg)
go c.importManager.flipTaskStateLoop(&c.wg)
}
// Start starts RootCoord.
func (c *Core) Start() error {
var err error