furyhawk/milvus
1
0
Fork 0
mirror of https://github.com/milvus-io/milvus.git synced 2026-07-21 10:15:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: clean up mlog call formatting (#51027)

Browse Source 
issue: #35917

- mlog formatting: remove blank lines inserted inside migrated log calls
- streaming logs: collapse short mlog calls in streaming
coordinator/node helpers for readability

Signed-off-by: chyezh <chyezh@outlook.com>
This commit is contained in:
Zhen Ye
2026-07-04 18:08:30 +08:00
committed by GitHub
parent 33f93c2aa1
commit e203c56f0a
22 changed files with 94 additions and 297 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/distributed/streaming/forward.go
+4 -12
View File
@@ -209,9 +209,7 @@ func (fs *forwardServiceImpl) initLegacyProxy() {
})
fs.legacyProxy = lazygrpc.WithServiceCreator(conn, milvuspb.NewMilvusServiceClient)
fs.rb = rb
fs.Logger().Info(context.TODO(),
"streaming service is not ready, legacy proxy is initiated to forward request", mlog.Int("proxyPort", port))
fs.Logger().Info(context.TODO(), "streaming service is not ready, legacy proxy is initiated to forward request", mlog.Int("proxyPort", port))
}
// getDialOptions returns the dial options for the legacy proxy.
@@ -255,9 +253,7 @@ func getDialOptions(rb resolver.Builder) []grpc.DialOption {
// markForwardDisabled marks the forward disabled.
func (fs *forwardServiceImpl) markForwardDisabled() {
fs.isForwardDisabled = true
fs.Logger().Info(context.TODO(),
"streaming service is ready, forward is disabled")
fs.Logger().Info(context.TODO(), "streaming service is ready, forward is disabled")
if fs.legacyProxy != nil {
legacyProxy := fs.legacyProxy
fs.legacyProxy = nil
@@ -265,13 +261,9 @@ func (fs *forwardServiceImpl) markForwardDisabled() {
fs.rb = nil
go func() {
legacyProxy.Close()
fs.Logger().Info(context.TODO(),
"legacy proxy closed")
fs.Logger().Info(context.TODO(), "legacy proxy closed")
rb.Close()
fs.Logger().Info(context.TODO(),
"resolver builder closed")
fs.Logger().Info(context.TODO(), "resolver builder closed")
}()
}
}
internal/distributed/streaming/internal/producer/producer.go
+4 -12
View File
@@ -155,9 +155,7 @@ func (p *ResumableProducer) produceInternal(ctx context.Context, msg message.Mut
// resumeLoop is used to resume producer from error.
func (p *ResumableProducer) resumeLoop() {
defer func() {
p.Logger().Info(p.ctx,
"stop resuming")
p.Logger().Info(p.ctx, "stop resuming")
p.metrics.IntoUnavailable()
close(p.resumingExitCh)
}()
@@ -188,9 +186,7 @@ func (p *ResumableProducer) waitUntilUnavailable(producer handler.Producer) erro
return p.ctx.Err()
case <-producer.Available():
// Wait old producer unavailable, trigger a new resuming operation.
p.Logger().Warn(p.ctx,
"producer encounter error, try to resume...")
p.Logger().Warn(p.ctx, "producer encounter error, try to resume...")
return nil
}
}
@@ -222,9 +218,7 @@ func (p *ResumableProducer) createNewProducer() (producer.Producer, error) {
// Otherwise, perform a resuming operation.
if err != nil {
nextBackoff := backoff.NextBackOff()
p.Logger().Warn(p.ctx,
"create producer failed, retry...", mlog.Err(err), mlog.Duration("nextRetryInterval", nextBackoff))
p.Logger().Warn(p.ctx, "create producer failed, retry...", mlog.Err(err), mlog.Duration("nextRetryInterval", nextBackoff))
time.Sleep(nextBackoff)
continue
}
@@ -250,9 +244,7 @@ func (p *ResumableProducer) gracefulClose() error {
// Close close the producer.
func (p *ResumableProducer) Close() {
if err := p.gracefulClose(); err != nil {
p.Logger().Warn(p.ctx,
"graceful close a producer fail, force close is applied")
p.Logger().Warn(p.ctx, "graceful close a producer fail, force close is applied")
}
// cancel is always need to be called, even graceful close is success.
internal/querycoordv2/load_config_watcher.go
+6 -22
View File
@@ -70,13 +70,9 @@ func (w *LoadConfigWatcher) Trigger() {
func (w *LoadConfigWatcher) background() {
defer func() {
w.notifier.Finish(struct{}{})
w.Logger().Info(context.TODO(),
"load config watcher stopped")
w.Logger().Info(context.TODO(), "load config watcher stopped")
}()
w.Logger().Info(context.TODO(),
"load config watcher started")
w.Logger().Info(context.TODO(), "load config watcher started")
balanceTimer := typeutil.NewBackoffTimer(typeutil.BackoffTimerConfig{
Default: loadConfigWatcherInterval,
@@ -95,9 +91,7 @@ func (w *LoadConfigWatcher) background() {
case <-w.notifier.Context().Done():
return
case <-w.triggerCh:
w.Logger().Info(context.TODO(),
"load config watcher triggered")
w.Logger().Info(context.TODO(), "load config watcher triggered")
case <-nextTimer:
case <-promotionTicker.C:
w.s.tryPromoteReadyLoadConfigReplicas(w.notifier.Context())
@@ -125,14 +119,12 @@ func (w *LoadConfigWatcher) applyLoadConfigChanges() error {
if newReplicaNum <= 0 || len(newRGs) == 0 {
w.Logger().Info(context.TODO(),
"illegal cluster level load config, skip it", mlog.Int32("replica_num", newReplicaNum), mlog.Strings("resource_groups", newRGs))
return nil
}
if len(newRGs) != 1 && len(newRGs) != int(newReplicaNum) {
w.Logger().Info(context.TODO(),
"illegal cluster level load config, skip it", mlog.Int32("replica_num", newReplicaNum), mlog.Strings("resource_groups", newRGs))
return nil
}
@@ -141,7 +133,6 @@ func (w *LoadConfigWatcher) applyLoadConfigChanges() error {
rgChanged := len(left) > 0 || len(right) > 0
if w.previousReplicaNum == newReplicaNum && !rgChanged {
w.Logger().Info(context.TODO(),
"no need to update load config, skip it", mlog.Int32("replica_num", newReplicaNum), mlog.Strings("resource_groups", newRGs))
return nil
}
@@ -151,29 +142,22 @@ func (w *LoadConfigWatcher) applyLoadConfigChanges() error {
collectionIDs = lo.Filter(collectionIDs, func(collectionID int64, _ int) bool {
collection := w.s.meta.GetCollection(w.notifier.Context(), collectionID)
if collection.UserSpecifiedReplicaMode {
w.Logger().Info(context.TODO(),
"collection is user specified replica mode, skip update load config", mlog.FieldCollectionID(collectionID))
w.Logger().Info(context.TODO(), "collection is user specified replica mode, skip update load config", mlog.FieldCollectionID(collectionID))
return false
}
return true
})
if len(collectionIDs) == 0 {
w.Logger().Info(context.TODO(),
"no collection to update load config, skip it")
w.Logger().Info(context.TODO(), "no collection to update load config, skip it")
}
if err := w.s.updateLoadConfig(w.notifier.Context(), collectionIDs, newReplicaNum, newRGs, true); err != nil {
w.Logger().Warn(context.TODO(),
"failed to update load config", mlog.Err(err))
w.Logger().Warn(context.TODO(), "failed to update load config", mlog.Err(err))
return err
}
w.s.tryPromoteReadyLoadConfigReplicas(w.notifier.Context())
w.Logger().Info(context.TODO(),
"apply load config changes",
mlog.Int64s("collectionIDs", collectionIDs),
mlog.Int32("previousReplicaNum", w.previousReplicaNum),
internal/rootcoord/tombstone/scheduler.go
+6 -18
View File
@@ -62,13 +62,9 @@ func (s *tombstoneSweeperImpl) AddTombstone(tombstone Tombstone) {
func (s *tombstoneSweeperImpl) background() {
defer func() {
s.notifier.Finish(struct{}{})
s.Logger().Info(context.TODO(),
"tombstone sweeper background exit")
s.Logger().Info(context.TODO(), "tombstone sweeper background exit")
}()
s.Logger().Info(context.TODO(),
"tombstone sweeper background start", mlog.Duration("interval", s.interval))
s.Logger().Info(context.TODO(), "tombstone sweeper background start", mlog.Duration("interval", s.interval))
ticker := time.NewTicker(s.interval)
defer ticker.Stop()
@@ -78,9 +74,7 @@ func (s *tombstoneSweeperImpl) background() {
case tombstone := <-s.incoming:
if _, ok := s.tombstones[tombstone.ID()]; !ok {
s.tombstones[tombstone.ID()] = tombstone
s.Logger().Info(context.TODO(),
"tombstone added", mlog.String("tombstone", tombstone.ID()))
s.Logger().Info(context.TODO(), "tombstone added", mlog.String("tombstone", tombstone.ID()))
}
case <-ticker.C:
s.triggerGCTombstone(s.notifier.Context())
@@ -103,24 +97,18 @@ func (s *tombstoneSweeperImpl) triggerGCTombstone(ctx context.Context) {
tombstoneID := tombstone.ID()
confirmed, err := tombstone.ConfirmCanBeRemoved(ctx)
if err != nil {
s.Logger().Warn(ctx,
"fail to confirm if tombstone can be removed", mlog.String("tombstone", tombstoneID), mlog.Err(err))
s.Logger().Warn(ctx, "fail to confirm if tombstone can be removed", mlog.String("tombstone", tombstoneID), mlog.Err(err))
continue
}
if !confirmed {
continue
}
if err := tombstone.Remove(ctx); err != nil {
s.Logger().Warn(ctx,
"fail to remove tombstone", mlog.String("tombstone", tombstoneID), mlog.Err(err))
s.Logger().Warn(ctx, "fail to remove tombstone", mlog.String("tombstone", tombstoneID), mlog.Err(err))
continue
}
delete(s.tombstones, tombstoneID)
s.Logger().Info(ctx,
"tombstone removed", mlog.String("tombstone", tombstoneID))
s.Logger().Info(ctx, "tombstone removed", mlog.String("tombstone", tombstoneID))
}
}
internal/streamingcoord/server/broadcaster/ack_callback_scheduler.go
+7 -25
View File
@@ -100,13 +100,9 @@ func (s *ackCallbackScheduler) Close() {
func (s *ackCallbackScheduler) background() {
defer func() {
s.notifier.Finish(struct{}{})
s.Logger().Info(context.TODO(),
"ack scheduler background exit")
s.Logger().Info(context.TODO(), "ack scheduler background exit")
}()
s.Logger().Info(context.TODO(),
"ack scheduler background start")
s.Logger().Info(context.TODO(), "ack scheduler background start")
// it's weired to find that FastLock may be failure even if there's no resource-key locked,
// also see: #45285
@@ -156,9 +152,7 @@ func (s *ackCallbackScheduler) triggerAckCallback() {
g, err := s.rkLocker.FastLock(task.Header().ResourceKeys.Collect()...)
if err != nil {
s.Logger().Warn(context.TODO(),
"lock is occupied, delay the ack callback", mlog.Uint64("broadcastID", task.Header().BroadcastID), mlog.Err(err))
s.Logger().Warn(context.TODO(), "lock is occupied, delay the ack callback", mlog.Uint64("broadcastID", task.Header().BroadcastID), mlog.Err(err))
pendingTasks = append(pendingTasks, task)
continue
}
@@ -205,16 +199,11 @@ func (s *ackCallbackScheduler) fixIncompleteBroadcastsForForcePromote(ctx contex
})
if len(incompleteTasks) == 0 {
s.Logger().Info(ctx,
"No incomplete broadcasts to fix for force promote")
s.Logger().Info(ctx, "No incomplete broadcasts to fix for force promote")
return nil
}
s.Logger().Info(ctx,
"Fixing incomplete broadcasts for force promote",
mlog.Int("incompleteTasks", len(incompleteTasks)))
s.Logger().Info(ctx, "Fixing incomplete broadcasts for force promote", mlog.Int("incompleteTasks", len(incompleteTasks)))
// Mark AlterReplicateConfig tasks with ignore=true (to prevent old config overwriting force promote config)
// MarkIgnore is a memory-only operation. It only runs on tasks where IsAlterReplicateConfigMessage() == true,
@@ -223,10 +212,7 @@ func (s *ackCallbackScheduler) fixIncompleteBroadcastsForForcePromote(ctx contex
if !task.IsAlterReplicateConfigMessage() {
continue
}
s.Logger().Info(ctx,
"Marking AlterReplicateConfig task with ignore=true",
mlog.Uint64("broadcastID", task.Header().BroadcastID))
s.Logger().Info(ctx, "Marking AlterReplicateConfig task with ignore=true", mlog.Uint64("broadcastID", task.Header().BroadcastID))
if err := task.MarkIgnore(); err != nil {
panic(fmt.Sprintf("unreachable: MarkIgnore failed on AlterReplicateConfig task %d: %v", task.Header().BroadcastID, err))
@@ -245,7 +231,6 @@ func (s *ackCallbackScheduler) fixIncompleteBroadcastsForForcePromote(ctx contex
pending.pendingMessages[i] = message.ClearReplicateHeader(msg)
}
s.Logger().Info(ctx,
"Delegating incomplete task to broadcastScheduler",
mlog.Uint64("broadcastID", task.Header().BroadcastID),
mlog.String("messageType", task.msg.MessageType().String()),
@@ -254,9 +239,7 @@ func (s *ackCallbackScheduler) fixIncompleteBroadcastsForForcePromote(ctx contex
return merr.Wrapf(err, "failed to supplement task %d via broadcastScheduler", task.Header().BroadcastID)
}
}
s.Logger().Info(ctx,
"All incomplete broadcasts fixed and tombstoned")
s.Logger().Info(ctx, "All incomplete broadcasts fixed and tombstoned")
return nil
}
@@ -321,7 +304,6 @@ func (s *ackCallbackScheduler) callMessageAckCallbackUntilDone(ctx context.Conte
}
nextInterval := backoff.NextBackOff()
s.Logger().Warn(ctx,
"failed to call message ack callback, wait for retry...",
mlog.FieldMessage(msg),
mlog.Duration("nextInterval", nextInterval),
internal/streamingcoord/server/broadcaster/broadcast_manager.go
+3 -11
View File
@@ -232,15 +232,12 @@ func (bm *broadcastTaskManager) LegacyAck(ctx context.Context, broadcastID uint6
task, ok := bm.getBroadcastTaskByID(broadcastID)
if !ok {
bm.Logger().Warn(ctx,
"broadcast task not found, it may already acked, ignore the request", mlog.Uint64("broadcastID", broadcastID), mlog.String("vchannel", vchannel))
return nil
}
msg := task.GetImmutableMessageFromVChannel(vchannel)
if msg == nil {
task.Logger().Warn(ctx,
"vchannel is already acked, ignore the ack request", mlog.String("vchannel", vchannel))
task.Logger().Warn(ctx, "vchannel is already acked, ignore the ack request", mlog.String("vchannel", vchannel))
return nil
}
return bm.Ack(ctx, msg)
@@ -256,7 +253,6 @@ func (bm *broadcastTaskManager) Ack(ctx context.Context, msg message.ImmutableMe
t, ok := bm.getOrCreateBroadcastTask(msg)
if !ok {
bm.Logger().Debug(ctx,
"task is tombstone, ignored the ack request",
mlog.Uint64("broadcastID", msg.BroadcastHeader().BroadcastID),
mlog.String("vchannel", msg.VChannel()))
@@ -274,9 +270,7 @@ func (bm *broadcastTaskManager) DropTombstone(ctx context.Context, broadcastID u
t, ok := bm.getBroadcastTaskByID(broadcastID)
if !ok {
bm.Logger().Debug(ctx,
"task is not found, ignored the drop tombstone request", mlog.Uint64("broadcastID", broadcastID))
bm.Logger().Debug(ctx, "task is not found, ignored the drop tombstone request", mlog.Uint64("broadcastID", broadcastID))
return nil
}
if err := t.DropTombstone(ctx); err != nil {
@@ -321,9 +315,7 @@ func (bm *broadcastTaskManager) getOrCreateBroadcastTask(msg message.ImmutableMe
return t, t.State() != streamingpb.BroadcastTaskState_BROADCAST_TASK_STATE_TOMBSTONE
}
if msg.ReplicateHeader() == nil {
bm.Logger().Warn(context.TODO(),
"try to recover task from the wal from non-replicate message, ignore it")
bm.Logger().Warn(context.TODO(), "try to recover task from the wal from non-replicate message, ignore it")
return nil, false
}
internal/streamingcoord/server/broadcaster/broadcast_scheduler.go
+3 -9
View File
@@ -73,15 +73,11 @@ func (b *broadcasterScheduler) execute() {
if workers < 1 {
workers = 1
}
b.Logger().Info(context.TODO(),
"broadcaster start to execute", mlog.Int("workerNum", workers))
b.Logger().Info(context.TODO(), "broadcaster start to execute", mlog.Int("workerNum", workers))
defer func() {
b.backgroundTaskNotifier.Finish(struct{}{})
b.Logger().Info(context.TODO(),
"broadcaster execute exit")
b.Logger().Info(context.TODO(), "broadcaster execute exit")
}()
// Start n workers to handle the broadcast task.
@@ -113,9 +109,7 @@ func (b *broadcasterScheduler) dispatch() {
if b.backoffs.Len() > 0 {
var nextInterval time.Duration
nextBackOff, nextInterval = b.backoffs.Peek().NextTimer()
b.Logger().Info(context.TODO(),
"backoff task", mlog.Duration("nextInterval", nextInterval))
b.Logger().Info(context.TODO(), "backoff task", mlog.Duration("nextInterval", nextInterval))
}
select {
internal/streamingcoord/server/broadcaster/tombstone_scheduler.go
+4 -12
View File
@@ -61,8 +61,7 @@ func (s *tombstoneScheduler) AddPending(broadcastID uint64) {
// must not panic. Dropping the in-memory enqueue is safe: the task state is
// already persisted as TOMBSTONE (MarkAckCallbackDone) before reaching here,
// and will be recovered into the GC list on the next startup.
s.Logger().Info(context.TODO(), "tombstone scheduler is closing, skip adding pending tombstone",
mlog.Uint64("broadcastID", broadcastID))
s.Logger().Info(context.TODO(), "tombstone scheduler is closing, skip adding pending tombstone", mlog.Uint64("broadcastID", broadcastID))
return
case s.pending <- broadcastID:
}
@@ -78,13 +77,9 @@ func (s *tombstoneScheduler) Close() {
func (s *tombstoneScheduler) background() {
defer func() {
s.notifier.Finish(struct{}{})
s.Logger().Info(context.TODO(),
"tombstone scheduler background exit")
s.Logger().Info(context.TODO(), "tombstone scheduler background exit")
}()
s.Logger().Info(context.TODO(),
"tombstone scheduler background start")
s.Logger().Info(context.TODO(), "tombstone scheduler background start")
tombstoneGCInterval := paramtable.Get().StreamingCfg.WALBroadcasterTombstoneCheckInternal.GetAsDurationByParse()
ticker := time.NewTicker(tombstoneGCInterval)
@@ -116,7 +111,6 @@ func (s *tombstoneScheduler) triggerGCTombstone() {
expiredOffset = len(s.tombstones) - maxTombstoneCount
}
s.Logger().Info(context.TODO(),
"triggerGCTombstone",
mlog.Int("tombstone count", len(s.tombstones)),
mlog.Int("expired offset", expiredOffset),
@@ -128,9 +122,7 @@ func (s *tombstoneScheduler) triggerGCTombstone() {
return
}
if err := s.bm.DropTombstone(s.notifier.Context(), tombstone.broadcastID); err != nil {
s.Logger().Error(context.TODO(),
"failed to drop tombstone", mlog.Err(err))
s.Logger().Error(context.TODO(), "failed to drop tombstone", mlog.Err(err))
s.tombstones = s.tombstones[idx:]
return
}
internal/streamingnode/server/service/handler/consumer/consume_server.go
+1 -3
View File
@@ -81,9 +81,7 @@ func CreateConsumeServer(walManager walmanager.Manager, streamServer streamingpb
}); err != nil {
// release the scanner to avoid resource leak.
if err := scanner.Close(); err != nil {
resource.Resource().Logger().Warn(context.TODO(),
"close scanner failed at create consume server", mlog.Err(err))
resource.Resource().Logger().Warn(context.TODO(), "close scanner failed at create consume server", mlog.Err(err))
}
return nil, err
}
internal/streamingnode/server/wal/adaptor/opener.go
+9 -26
View File
@@ -368,9 +368,7 @@ func (o *openerAdaptorImpl) handleAlterWALFlushingStage(ctx context.Context, opt
mlog.Duration("timeout", defaultWALSwitchFlushTimeout))
return status.NewInner("timeout waiting for flush completion during WAL switch")
case <-ctx.Done():
mlog.Warn(ctx, "context canceled while waiting for flush completion",
mlog.String("channel", opt.Channel.Name),
mlog.Err(ctx.Err()))
mlog.Warn(ctx, "context canceled while waiting for flush completion", mlog.String("channel", opt.Channel.Name), mlog.Err(ctx.Err()))
return errors.Wrap(ctx.Err(), "context canceled during WAL switch flush waiting")
}
}
@@ -389,9 +387,7 @@ func (o *openerAdaptorImpl) handleAlterWALFlushingStage(ctx context.Context, opt
catalog := resource.Resource().StreamingNodeCatalog()
if err := catalog.SaveConsumeCheckpoint(ctx, opt.Channel.Name, snapshot.Checkpoint.IntoProto()); err != nil {
mlog.Warn(ctx, "failed to persist checkpoint after flushing stage",
mlog.String("channel", opt.Channel.Name),
mlog.Err(err))
mlog.Warn(ctx, "failed to persist checkpoint after flushing stage", mlog.String("channel", opt.Channel.Name), mlog.Err(err))
return errors.Wrap(err, "failed to persist checkpoint after flushing stage")
}
@@ -461,9 +457,7 @@ func (o *openerAdaptorImpl) handleAlterWALAdvanceCheckpointsStage(ctx context.Co
for _, vchannel := range vchannels {
resp2, err2 := mixCoordClient.GetChannelRecoveryInfo(ctx, &datapb.GetChannelRecoveryInfoRequest{Vchannel: vchannel.Vchannel})
if err2 != nil {
mlog.Warn(ctx, "failed to verify vchannel checkpoint update",
mlog.String("vchannel", vchannel.Vchannel),
mlog.Err(err2))
mlog.Warn(ctx, "failed to verify vchannel checkpoint update", mlog.String("vchannel", vchannel.Vchannel), mlog.Err(err2))
return errors.Wrap(err2, "failed to verify vchannel checkpoint update")
}
mlog.Info(ctx, "verified vchannel checkpoint update",
@@ -471,8 +465,7 @@ func (o *openerAdaptorImpl) handleAlterWALAdvanceCheckpointsStage(ctx context.Co
mlog.Binary("seekPositionMsgID", resp2.Info.SeekPosition.MsgID))
}
} else {
mlog.Info(ctx, "no vchannels found, skipping vchannel checkpoint update",
mlog.String("channel", opt.Channel.Name))
mlog.Info(ctx, "no vchannels found, skipping vchannel checkpoint update", mlog.String("channel", opt.Channel.Name))
}
// Update pchannel checkpoint: reset alterWALState and set position to new WAL initial position
@@ -485,9 +478,7 @@ func (o *openerAdaptorImpl) handleAlterWALAdvanceCheckpointsStage(ctx context.Co
// Persist final checkpoint to catalog
if err := catalog.SaveConsumeCheckpoint(ctx, opt.Channel.Name, finalCheckpoint.IntoProto()); err != nil {
mlog.Warn(ctx, "failed to persist checkpoint after advance checkpoint stage",
mlog.String("channel", opt.Channel.Name),
mlog.Err(err))
mlog.Warn(ctx, "failed to persist checkpoint after advance checkpoint stage", mlog.String("channel", opt.Channel.Name), mlog.Err(err))
return errors.Wrap(err, "failed to persist checkpoint after advance checkpoint stage")
}
@@ -518,16 +509,12 @@ func (o *openerAdaptorImpl) Close() {
o.lifetime.SetState(typeutil.LifetimeStateStopped)
o.lifetime.Wait()
o.Logger().Info(context.TODO(),
"wal opener closing...")
o.Logger().Info(context.TODO(), "wal opener closing...")
// close all wal instances.
o.walInstances.Range(func(id int64, l wal.WAL) bool {
l.Close()
o.Logger().Info(context.TODO(),
"close wal by opener", mlog.Int64("id", id), mlog.String("channel", l.Channel().String()))
o.Logger().Info(context.TODO(), "close wal by opener", mlog.Int64("id", id), mlog.String("channel", l.Channel().String()))
return true
})
@@ -535,14 +522,10 @@ func (o *openerAdaptorImpl) Close() {
o.mu.Lock()
defer o.mu.Unlock()
for walName, opener := range o.openerCache {
o.Logger().Info(context.TODO(),
"closing underlying walimpls opener", mlog.Stringer("walName", walName))
o.Logger().Info(context.TODO(), "closing underlying walimpls opener", mlog.Stringer("walName", walName))
opener.Close()
}
o.openerCache = nil
o.Logger().Info(context.TODO(),
"wal opener closed")
o.Logger().Info(context.TODO(), "wal opener closed")
}
internal/streamingnode/server/wal/adaptor/ro_wal_adaptor.go
+5 -16
View File
@@ -123,7 +123,6 @@ func (w *roWALAdaptorImpl) checkReadOptWALName(opts wal.ReadOption) error {
msgWALName := message.WALName(msgID.WALName)
if msgWALName != currentWALName {
w.Logger().Info(context.TODO(),
"WAL name mismatch", mlog.String("msgIDWALName", msgWALName.String()), mlog.String("currentWALName", currentWALName.String()))
return status.NewWALNameMismatchError(currentWALName.String(), msgWALName.String())
}
@@ -145,16 +144,12 @@ func (w *roWALAdaptorImpl) Available() <-chan struct{} {
// Close overrides Scanner Close function.
func (w *roWALAdaptorImpl) Close() {
// begin to close the wal.
w.Logger().Info(context.TODO(),
"wal begin to close...")
w.Logger().Info(context.TODO(), "wal begin to close...")
w.lifetime.SetState(typeutil.LifetimeStateStopped)
w.forceCancelAfterGracefulTimeout()
w.lifetime.Wait()
w.Logger().Info(context.TODO(),
"wal begin to close scanners...")
w.Logger().Info(context.TODO(), "wal begin to close scanners...")
// close all wal instances.
w.scanners.Range(func(id int64, s wal.Scanner) bool {
@@ -163,18 +158,12 @@ func (w *roWALAdaptorImpl) Close() {
return true
})
w.Logger().Info(context.TODO(),
"scanner close done, close inner wal...")
w.Logger().Info(context.TODO(), "scanner close done, close inner wal...")
w.roWALImpls.Close()
w.Logger().Info(context.TODO(),
"call wal cleanup function...")
w.Logger().Info(context.TODO(), "call wal cleanup function...")
w.cleanup()
w.Logger().Info(context.TODO(),
"wal closed")
w.Logger().Info(context.TODO(), "wal closed")
// close all metrics.
w.scanMetrics.Close()
internal/streamingnode/server/wal/interceptors/shard/shards/segment_alloc_worker.go
+7 -22
View File
@@ -19,9 +19,7 @@ import (
// asyncAllocSegment allocates a new growing segment asynchronously.
func (m *partitionManager) asyncAllocSegment(schemaVersion int32, useGrowingSourceFlush bool) {
if m.onAllocating != nil {
m.Logger().Debug(context.TODO(),
"segment alloc worker is already on allocating")
m.Logger().Debug(context.TODO(), "segment alloc worker is already on allocating")
// manager is already on allocating.
return
}
@@ -73,26 +71,18 @@ func (w *segmentAllocWorker) do() {
return
}
if e := status.AsStreamingError(err); e.IsUnrecoverable() {
w.Logger().Warn(w.ctx,
"allocate new growing segement with unrecoverable error, stop retrying", mlog.Err(err))
w.Logger().Warn(w.ctx, "allocate new growing segement with unrecoverable error, stop retrying", mlog.Err(err))
return
}
nextInterval := backoff.NextBackOff()
w.Logger().Info(w.ctx,
"failed to allocate new growing segment, retrying", mlog.Duration("nextInterval", nextInterval), mlog.Err(err))
w.Logger().Info(w.ctx, "failed to allocate new growing segment, retrying", mlog.Duration("nextInterval", nextInterval), mlog.Err(err))
select {
case <-w.ctx.Done():
w.Logger().Info(w.ctx,
"segment allocation canceled", mlog.Err(w.ctx.Err()))
w.Logger().Info(w.ctx, "segment allocation canceled", mlog.Err(w.ctx.Err()))
return
case <-w.wal.Available():
// wal is unavailable, stop the worker.
w.Logger().Warn(w.ctx,
"wal is unavailable, stop alloc new segment")
w.Logger().Warn(w.ctx, "wal is unavailable, stop alloc new segment")
return
case <-time.After(backoff.NextBackOff()):
}
@@ -131,13 +121,10 @@ func (w *segmentAllocWorker) doOnce() error {
result, err := w.wal.Append(w.ctx, msg)
if err != nil {
w.Logger().Warn(w.ctx,
"failed to append create segment message", mlog.FieldMessage(msg), mlog.Err(err))
w.Logger().Warn(w.ctx, "failed to append create segment message", mlog.FieldMessage(msg), mlog.Err(err))
return err
}
w.Logger().Info(w.ctx,
"append create segment message", mlog.FieldMessage(msg), mlog.String("messageID", result.MessageID.String()), mlog.Uint64("timetick", result.TimeTick))
return nil
}
@@ -153,9 +140,7 @@ func (w *segmentAllocWorker) initSegmentConfig() error {
// Allocate new segment id.
segmentID, err := resource.Resource().IDAllocator().Allocate(w.ctx)
if err != nil {
w.Logger().Warn(w.ctx,
"failed to allocate segment id", mlog.Err(err))
w.Logger().Warn(w.ctx, "failed to allocate segment id", mlog.Err(err))
return err
}
w.segmentID = segmentID
internal/streamingnode/server/wal/interceptors/shard/shards/segment_flush_worker.go
+9 -28
View File
@@ -58,9 +58,7 @@ func (w *segmentFlushWorker) do() {
// Otherwise, the flush message may be sent into wal before the txn is done.
// Break the wal consistency: All insert message is written into wal before the flush message.
if err := w.waitForTxnManagerRecoverDone(); err != nil {
w.Logger().Error(w.ctx,
"failed to wait for txn manager recover ready", mlog.Err(err))
w.Logger().Error(w.ctx, "failed to wait for txn manager recover ready", mlog.Err(err))
return
}
@@ -70,27 +68,19 @@ func (w *segmentFlushWorker) do() {
return
}
if e := status.AsStreamingError(err); e.IsUnrecoverable() {
w.Logger().Warn(w.ctx,
"flush growing segement with unrecoverable error, stop retrying", mlog.Err(err))
w.Logger().Warn(w.ctx, "flush growing segement with unrecoverable error, stop retrying", mlog.Err(err))
return
}
nextInterval := backoff.NextBackOff()
w.Logger().Info(w.ctx,
"failed to flush new growing segment, retrying", mlog.Duration("nextInterval", nextInterval), mlog.Err(err))
w.Logger().Info(w.ctx, "failed to flush new growing segment, retrying", mlog.Duration("nextInterval", nextInterval), mlog.Err(err))
select {
case <-w.ctx.Done():
w.Logger().Info(w.ctx,
"flush segment canceled", mlog.Err(w.ctx.Err()))
w.Logger().Info(w.ctx, "flush segment canceled", mlog.Err(w.ctx.Err()))
return
case <-w.wal.Available():
// wal is unavailable, stop the worker.
w.Logger().Warn(w.ctx,
"wal is unavailable, stop flush segment")
w.Logger().Warn(w.ctx, "wal is unavailable, stop flush segment")
return
case <-time.After(backoff.NextBackOff()):
}
@@ -104,9 +94,7 @@ func (w *segmentFlushWorker) waitForTxnManagerRecoverDone() error {
// txn manager is ready, continue to do the flush.
return nil
case <-w.ctx.Done():
w.Logger().Info(w.ctx,
"flush segment canceled", mlog.Err(w.ctx.Err()))
w.Logger().Info(w.ctx, "flush segment canceled", mlog.Err(w.ctx.Err()))
return w.ctx.Err()
case <-w.wal.Available():
return status.NewOnShutdownError("wal is unavailable")
@@ -133,14 +121,11 @@ func (w *segmentFlushWorker) doOnce() error {
result, err := w.wal.Append(w.ctx, msg)
if err != nil {
w.Logger().Error(w.ctx,
"failed to append flush message", mlog.FieldMessage(msg), mlog.Err(err))
w.Logger().Error(w.ctx, "failed to append flush message", mlog.FieldMessage(msg), mlog.Err(err))
return err
}
policy := w.segment.SealPolicy()
w.Logger().Info(w.ctx,
"segment has been flushed",
mlog.FieldMessage(msg),
mlog.String("policy", string(policy.Policy)),
@@ -155,16 +140,12 @@ func (w *segmentFlushWorker) doOnce() error {
func (w *segmentFlushWorker) checkIfReady() bool {
// if there're flying acks, wait them acked, delay the flush at next retry.
if ackSem := w.segment.AckSem(); ackSem > 0 {
w.Logger().Info(w.ctx,
"segment has flying insert operation, delay it", mlog.Int32("ackSem", ackSem), mlog.FieldSegmentID(w.segment.GetSegmentID()))
w.Logger().Info(w.ctx, "segment has flying insert operation, delay it", mlog.Int32("ackSem", ackSem), mlog.FieldSegmentID(w.segment.GetSegmentID()))
return false
}
// if there're flying txns, wait them committed, delay the flush at next retry.
if txnSem := w.segment.TxnSem(); txnSem > 0 {
w.Logger().Info(w.ctx,
"segment has flying txns, delay it", mlog.Int32("txnSem", txnSem), mlog.FieldSegmentID(w.segment.GetSegmentID()))
w.Logger().Info(w.ctx, "segment has flying txns, delay it", mlog.Int32("txnSem", txnSem), mlog.FieldSegmentID(w.segment.GetSegmentID()))
return false
}
return true
internal/streamingnode/server/wal/interceptors/shard/stats/stats_manager.go
+1 -3
View File
@@ -518,9 +518,7 @@ func (m *StatsManager) updateConfig() {
m.mu.Lock()
defer m.mu.Unlock()
if m.cfg != cfg {
m.Logger().Info(context.TODO(),
"update stats manager config", mlog.Any("newConfig", cfg), mlog.Any("oldConfig", m.cfg))
m.Logger().Info(context.TODO(), "update stats manager config", mlog.Any("newConfig", cfg), mlog.Any("oldConfig", m.cfg))
m.cfg = cfg
}
}
internal/streamingnode/server/wal/interceptors/shard/stats/stats_seal_worker.go
+2 -7
View File
@@ -96,9 +96,7 @@ func (m *sealWorker) loop() {
func (m *sealWorker) notifyToSealSegmentWithTimePolicy() {
sealSegmentIDs := m.statsManager.selectSegmentsWithTimePolicy()
if len(sealSegmentIDs) != 0 {
m.Logger().Info(context.TODO(),
"notify to seal segments with time policy", mlog.Int("segmentNum", len(sealSegmentIDs)))
m.Logger().Info(context.TODO(), "notify to seal segments with time policy", mlog.Int("segmentNum", len(sealSegmentIDs)))
for segmentID, sealPolicy := range sealSegmentIDs {
m.asyncMustSealSegment(segmentID, sealPolicy)
}
@@ -109,9 +107,7 @@ func (m *sealWorker) notifyToSealSegmentWithTimePolicy() {
func (m *sealWorker) notifyToSealSegmentWithBlockingL0Policy() {
sealSegmentIDs := m.statsManager.selectSegmentsWithBlockingL0Policy()
if len(sealSegmentIDs) != 0 {
m.Logger().Info(context.TODO(),
"notify to seal segments with blocking l0 policy", mlog.Int("segmentNum", len(sealSegmentIDs)))
m.Logger().Info(context.TODO(), "notify to seal segments with blocking l0 policy", mlog.Int("segmentNum", len(sealSegmentIDs)))
for segmentID, sealPolicy := range sealSegmentIDs {
m.asyncMustSealSegment(segmentID, sealPolicy)
}
@@ -123,7 +119,6 @@ func (m *sealWorker) notifyToSealSegmentUntilLessThanLWM(sealPolicy policy.SealP
segmentIDs := m.statsManager.selectSegmentsUntilLessThanLWM()
if len(segmentIDs) != 0 {
m.Logger().Info(context.TODO(),
"notify to seal segments until less than LWM", mlog.Int("segmentNum", len(segmentIDs)), mlog.String("policy", string(sealPolicy.Policy)))
for _, segmentID := range segmentIDs {
m.asyncMustSealSegment(segmentID, sealPolicy)
internal/streamingnode/server/wal/interceptors/txn/txn_manager.go
+6 -19
View File
@@ -48,9 +48,7 @@ func NewTxnManager(pchannel types.PChannelInfo, uncommittedTxnBuilders map[messa
}
txnManager.notifyRecoverDone()
txnManager.SetLogger(resource.Resource().Logger().With(mlog.FieldComponent("txn-manager")))
txnManager.Logger().Info(context.TODO(),
"txn manager recovered with txn", mlog.Int64s("txnIDs", sessionIDs))
txnManager.Logger().Info(context.TODO(), "txn manager recovered with txn", mlog.Int64s("txnIDs", sessionIDs))
return txnManager
}
@@ -142,9 +140,7 @@ func (m *TxnManager) FailTxnAtVChannel(vchannel string) {
}
}
if len(ids) > 0 {
m.Logger().Info(context.TODO(),
"transaction interrupted", mlog.FieldVChannel(vchannel), mlog.Int64s("txnIDs", ids))
m.Logger().Info(context.TODO(), "transaction interrupted", mlog.FieldVChannel(vchannel), mlog.Int64s("txnIDs", ids))
}
m.notifyRecoverDone()
}
@@ -197,15 +193,11 @@ func (m *TxnManager) RollbackAllInFlightTransactions() {
defer m.mu.Unlock()
if len(m.sessions) == 0 {
m.Logger().Info(context.TODO(),
"No in-flight transactions to rollback")
m.Logger().Info(context.TODO(), "No in-flight transactions to rollback")
return
}
m.Logger().Info(context.TODO(),
"Rolling back all in-flight transactions", mlog.Int("sessionCount", len(m.sessions)))
m.Logger().Info(context.TODO(), "Rolling back all in-flight transactions", mlog.Int("sessionCount", len(m.sessions)))
ids := make([]int64, 0, len(m.sessions))
for txnID, session := range m.sessions {
@@ -215,10 +207,7 @@ func (m *TxnManager) RollbackAllInFlightTransactions() {
delete(m.recoveredSessions, txnID)
}
m.Logger().Info(context.TODO(),
"Rolled back in-flight transactions",
mlog.Int64s("txnIDs", ids))
m.Logger().Info(context.TODO(), "Rolled back in-flight transactions", mlog.Int64s("txnIDs", ids))
// Signal GracefulClose if it's already waiting and all sessions are now cleared.
if len(m.sessions) == 0 && m.closed != nil {
@@ -239,9 +228,7 @@ func (m *TxnManager) GracefulClose(ctx context.Context) error {
m.closed.Close()
}
}
m.Logger().Info(ctx,
"graceful close txn manager", mlog.Int("activeTxnCount", len(m.sessions)))
m.Logger().Info(ctx, "graceful close txn manager", mlog.Int("activeTxnCount", len(m.sessions)))
m.mu.Unlock()
select {
internal/streamingnode/server/wal/metricsutil/wal_write.go
+2 -6
View File
@@ -96,16 +96,12 @@ func (m *WriteMetrics) done(appendMetrics *AppendMetrics) {
}
}
if appendMetrics.err != nil {
m.Logger().Warn(context.TODO(),
"append message into wal failed", appendMetrics.IntoLogFields()...)
m.Logger().Warn(context.TODO(), "append message into wal failed", appendMetrics.IntoLogFields()...)
return
}
if appendMetrics.appendDuration >= m.slowLogThreshold {
// log slow append catch
m.Logger().Warn(context.TODO(),
"append message into wal too slow", appendMetrics.IntoLogFields()...)
m.Logger().Warn(context.TODO(), "append message into wal too slow", appendMetrics.IntoLogFields()...)
return
}
logLV := appendMetrics.msg.MessageType().LogLevel()
internal/streamingnode/server/wal/recovery/recovery_background_task.go
+3 -9
View File
@@ -36,18 +36,12 @@ func (rs *recoveryStorageImpl) backgroundTask() {
ticker := time.NewTicker(rs.cfg.persistInterval)
defer func() {
ticker.Stop()
rs.Logger().Info(context.TODO(),
"recovery storage background task, perform a graceful exit...")
rs.Logger().Info(context.TODO(), "recovery storage background task, perform a graceful exit...")
if err := rs.persistDritySnapshotWhenClosing(); err != nil {
rs.Logger().Warn(context.TODO(),
"failed to persist dirty snapshot when closing", mlog.Err(err))
rs.Logger().Warn(context.TODO(), "failed to persist dirty snapshot when closing", mlog.Err(err))
}
rs.backgroundTaskNotifier.Finish(struct{}{})
rs.Logger().Info(context.TODO(),
"recovery storage background task exit")
rs.Logger().Info(context.TODO(), "recovery storage background task exit")
}()
for {
internal/streamingnode/server/walmanager/manager_impl.go
+1 -3
View File
@@ -23,9 +23,7 @@ var errWALManagerClosed = status.NewOnShutdownError("wal manager is closed")
// OpenManager create a WAL Manager, which now uses dynamic opener that can handle multiple WALNames at runtime.
// The specific WALName will be determined when opening each channel based on checkpoint's MessageID.WALName
func OpenManager() (Manager, error) {
resource.Resource().Logger().Info(context.TODO(),
"open wal manager with dynamic opener")
resource.Resource().Logger().Info(context.TODO(), "open wal manager with dynamic opener")
// Create dynamic opener directly with interceptors
opener := adaptor.NewOpenerAdaptor(
[]interceptors.InterceptorBuilder{
internal/util/streamingutil/service/resolver/resolver_with_discoverer.go
+9 -28
View File
@@ -138,30 +138,22 @@ func (r *resolverWithDiscoverer) doDiscover() {
// Check if all grpc resolver is stopped.
for r := range grpcResolvers {
if r.State() == typeutil.LifetimeStateWorking {
r.Logger().Warn(context.TODO(),
"resolver is stopped before grpc watcher exist, maybe bug here")
r.Logger().Warn(context.TODO(), "resolver is stopped before grpc watcher exist, maybe bug here")
break
}
}
r.Logger().Info(context.TODO(),
"resolver stopped")
r.Logger().Info(context.TODO(), "resolver stopped")
r.taskNotifier.Finish(struct{}{})
}()
for {
ch := r.asyncDiscover(r.taskNotifier.Context())
r.Logger().Info(context.TODO(),
"service discover task started, listening...")
r.Logger().Info(context.TODO(), "service discover task started, listening...")
L:
for {
select {
case watcher := <-r.registerCh:
watcher.Logger().Info(context.TODO(),
"new grpc resolver registered")
watcher.Logger().Info(context.TODO(), "new grpc resolver registered")
// New grpc resolver registered.
// Trigger the latest state to the new grpc resolver.
grpcResolvers[watcher] = struct{}{}
@@ -170,9 +162,7 @@ func (r *resolverWithDiscoverer) doDiscover() {
continue
}
if err := watcher.Update(*state); err != nil {
r.Logger().Info(context.TODO(),
"resolver is closed, ignore the new grpc resolver", mlog.Err(err))
r.Logger().Info(context.TODO(), "resolver is closed, ignore the new grpc resolver", mlog.Err(err))
delete(grpcResolvers, watcher)
}
case stateWithError := <-ch:
@@ -181,9 +171,7 @@ func (r *resolverWithDiscoverer) doDiscover() {
// resolver stopped.
return
}
r.Logger().Warn(context.TODO(),
"service discover break down", mlog.Err(stateWithError.err), mlog.Duration("retryInterval", r.retryInterval))
r.Logger().Warn(context.TODO(), "service discover break down", mlog.Err(stateWithError.err), mlog.Duration("retryInterval", r.retryInterval))
time.Sleep(r.retryInterval)
break L
}
@@ -193,29 +181,22 @@ func (r *resolverWithDiscoverer) doDiscover() {
latestState := r.getLatestState()
if latestState != nil && !state.Version.GT(latestState.Version) {
// Ignore the old version.
r.Logger().Info(context.TODO(),
"service discover update, ignore old version", mlog.Stringer("state", state))
r.Logger().Info(context.TODO(), "service discover update, ignore old version", mlog.Stringer("state", state))
continue
}
// Update all grpc resolver.
r.Logger().Info(context.TODO(),
"service discover update, update resolver", mlog.Stringer("state", state), mlog.Int("resolver_count", len(grpcResolvers)))
for watcher := range grpcResolvers {
// Update operation do not block.
// Only return error if the resolver is closed, so just print a info log and delete the resolver.
if err := watcher.Update(state); err != nil {
// updateError is always context.Canceled.
r.Logger().Info(context.TODO(),
"resolver is closed, unregister the resolver", mlog.NamedError("updateError", err))
r.Logger().Info(context.TODO(), "resolver is closed, unregister the resolver", mlog.NamedError("updateError", err))
delete(grpcResolvers, watcher)
}
}
r.Logger().Info(context.TODO(),
"update resolver done")
r.Logger().Info(context.TODO(), "update resolver done")
// Update the latest state and notify all resolver watcher should be executed after the all grpc watcher updated.
r.latestStateCond.LockAndBroadcast()
r.latestState = &state
pkg/kv/reliable_write_meta_kv.go
+1 -3
View File
@@ -97,9 +97,7 @@ func (kv *ReliableWriteMetaKv) retryWithBackoff(ctx context.Context, fn func(ctx
case <-ctx.Done():
return ctx.Err()
case <-time.After(nextInterval):
kv.Logger().Warn(ctx,
"failed to persist operation, wait for retry...", mlog.Duration("nextRetryInterval", nextInterval), mlog.Err(err))
kv.Logger().Warn(ctx, "failed to persist operation, wait for retry...", mlog.Duration("nextRetryInterval", nextInterval), mlog.Err(err))
}
}
}
pkg/streaming/walimpls/impls/pulsar/backlog_clear_helper.go
+1 -3
View File
@@ -134,9 +134,7 @@ func (h *backlogClearHelper) closeConsumer() {
if h.reusedConsumer != nil {
h.reusedConsumer.Close()
h.reusedConsumer = nil
h.Logger().Info(context.TODO(),
"closed the reused consumer")
h.Logger().Info(context.TODO(), "closed the reused consumer")
}
}