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

Fix the misleading log messages (#22606)

Browse Source 
Signed-off-by: yangxuan <xuan.yang@zilliz.com>
This commit is contained in:
XuanYang-cn 2023-03-09 10:39:51 +08:00 committed by GitHub
parent 11f1f4226a
commit b3287ca5ec
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 21 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
internal/querynode/load_segment_task.go
View File

@ -18,7 +18,6 @@ package querynode
import ( import (
"context" "context"
"fmt"
"go.uber.org/zap" "go.uber.org/zap"
"golang.org/x/sync/errgroup" "golang.org/x/sync/errgroup"
@ -133,18 +132,21 @@ func (l *loadSegmentsTask) Execute(ctx context.Context) error {
} }
// internal helper function to subscribe delta channel // internal helper function to subscribe delta channel
func (l *loadSegmentsTask) watchDeltaChannel(deltaChannels []string) error { func (l *loadSegmentsTask) watchDeltaChannel(dmlChannels []string) error {
collectionID := l.req.CollectionID var (
collectionID = l.req.CollectionID
vDeltaChannels []string
VPDeltaChannels = make(map[string]string)
)
log := log.With( log := log.With(
zap.Int64("collectionID", collectionID), zap.Int64("collectionID", collectionID),
zap.Strings("dmlChannels", dmlChannels),
) )
var vDeltaChannels []string for _, v := range dmlChannels {
VPDeltaChannels := make(map[string]string)
for _, v := range deltaChannels {
dc, err := funcutil.ConvertChannelName(v, Params.CommonCfg.RootCoordDml.GetValue(), Params.CommonCfg.RootCoordDelta.GetValue()) dc, err := funcutil.ConvertChannelName(v, Params.CommonCfg.RootCoordDml.GetValue(), Params.CommonCfg.RootCoordDelta.GetValue())
if err != nil { if err != nil {
log.Warn("watchDeltaChannels, failed to convert deltaChannel from dmlChannel", log.Warn("watchDeltaChannels, failed to convert deltaChannel from dmlChannel",
zap.String("DmlChannel", v), zap.String("dmlChannel", v),
zap.Error(err), zap.Error(err),
) )
return err return err
@ -153,9 +155,7 @@ func (l *loadSegmentsTask) watchDeltaChannel(deltaChannels []string) error {
vDeltaChannels = append(vDeltaChannels, dc) vDeltaChannels = append(vDeltaChannels, dc)
VPDeltaChannels[dc] = p VPDeltaChannels[dc] = p
} }
log.Info("Starting WatchDeltaChannels ...", log.Info("Starting WatchDeltaChannels ...", zap.Strings("deltaChannels", vDeltaChannels))
zap.Strings("channels", vDeltaChannels),
)
coll, err := l.node.metaReplica.getCollectionByID(collectionID) coll, err := l.node.metaReplica.getCollectionByID(collectionID)
if err != nil { if err != nil {
@ -165,14 +165,14 @@ func (l *loadSegmentsTask) watchDeltaChannel(deltaChannels []string) error {
// add collection meta and fg with mutex protection. // add collection meta and fg with mutex protection.
channel2FlowGraph, err := l.node.dataSyncService.addFlowGraphsForDeltaChannels(collectionID, vDeltaChannels, VPDeltaChannels) channel2FlowGraph, err := l.node.dataSyncService.addFlowGraphsForDeltaChannels(collectionID, vDeltaChannels, VPDeltaChannels)
if err != nil { if err != nil {
log.Warn("watchDeltaChannel, add flowGraph for deltaChannel failed", zap.Int64("collectionID", collectionID), zap.Strings("vDeltaChannels", vDeltaChannels), zap.Error(err)) log.Warn("watchDeltaChannel, failed to add flowGraph for deltaChannels",
zap.Strings("deltaChannels", vDeltaChannels),
zap.Error(err))
return err return err
} }
if len(channel2FlowGraph) == 0 { if len(channel2FlowGraph) == 0 {
log.Warn("all delta channels have been added before", log.Warn("all delta channels have been added before", zap.Strings("deltaChannels", vDeltaChannels))
zap.Strings("deltaChannels", deltaChannels),
)
return nil return nil
} }
@ -189,7 +189,7 @@ func (l *loadSegmentsTask) watchDeltaChannel(deltaChannels []string) error {
} }
}() }()
log.Info("watchDeltaChannel, add flowGraph for deltaChannel success", zap.Int64("collectionID", collectionID), zap.Strings("vDeltaChannels", vDeltaChannels)) log.Info("watchDeltaChannel, add flowGraph for deltaChannel success", zap.Strings("deltaChannels", vDeltaChannels))
// create tSafe // create tSafe
for _, channel := range vDeltaChannels { for _, channel := range vDeltaChannels {
@ -200,12 +200,15 @@ func (l *loadSegmentsTask) watchDeltaChannel(deltaChannels []string) error {
for _, channel := range vDeltaChannels { for _, channel := range vDeltaChannels {
dmlChannel, err := funcutil.ConvertChannelName(channel, Params.CommonCfg.RootCoordDelta.GetValue(), Params.CommonCfg.RootCoordDml.GetValue()) dmlChannel, err := funcutil.ConvertChannelName(channel, Params.CommonCfg.RootCoordDelta.GetValue(), Params.CommonCfg.RootCoordDml.GetValue())
if err != nil { if err != nil {
log.Error("failed to convert delta channel to dml", zap.String("channel", channel), zap.Error(err)) log.Error("failed to convert delta channel to dml", zap.String("deltaChannel", channel), zap.Error(err))
panic(err) panic(err)
} }
err = l.node.queryShardService.addQueryShard(collectionID, dmlChannel, l.req.GetReplicaID()) err = l.node.queryShardService.addQueryShard(collectionID, dmlChannel, l.req.GetReplicaID())
if err != nil { if err != nil {
log.Error("failed to add shard Service to query shard", zap.String("channel", channel), zap.Error(err)) log.Error("failed to add shard Service to query shard",
zap.String("dmlChannel", dmlChannel),
zap.String("deltaChannel", channel),
zap.Error(err))
panic(err) panic(err)
} }
} }
@ -215,6 +218,6 @@ func (l *loadSegmentsTask) watchDeltaChannel(deltaChannels []string) error {
fg.flowGraph.Start() fg.flowGraph.Start()
} }
log.Info("WatchDeltaChannels done", zap.Int64("collectionID", collectionID), zap.String("ChannelIDs", fmt.Sprintln(vDeltaChannels))) log.Info("WatchDeltaChannels done", zap.Strings("deltaChannels", vDeltaChannels))
return nil return nil
} }