From d1fc17ad0df3b10d5b192a8dbe177f93ed55ea42 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Tue, 14 Apr 2026 17:04:53 +0800 Subject: [PATCH 1/4] remove some verbose and useless logs --- cdc/kv/shared_client.go | 6 +----- .../internal/v3/transport/transport.go | 7 ------- .../dmlsink/cloudstorage/encoding_worker.go | 6 ------ pkg/config/sink.go | 5 ----- pkg/txnutil/lock_resolver.go | 21 ------------------- 5 files changed, 1 insertion(+), 44 deletions(-) diff --git a/cdc/kv/shared_client.go b/cdc/kv/shared_client.go index 6953cbc32a..c81c9f40d0 100644 --- a/cdc/kv/shared_client.go +++ b/cdc/kv/shared_client.go @@ -824,7 +824,7 @@ func (s *SharedClient) handleResolveLockTasks(ctx context.Context) error { } func (s *SharedClient) logSlowRegions(ctx context.Context) error { - ticker := time.NewTicker(30 * time.Second) + ticker := time.NewTicker(5 * time.Minute) defer ticker.Stop() for { select { @@ -832,10 +832,6 @@ func (s *SharedClient) logSlowRegions(ctx context.Context) error { return ctx.Err() case <-ticker.C: } - log.Info("event feed starts to check locked regions", - zap.String("namespace", s.changefeed.Namespace), - zap.String("changefeed", s.changefeed.ID)) - currTime := s.pdClock.CurrentTime() s.totalSpans.RLock() var slowInitializeRegionCount int diff --git a/cdc/scheduler/internal/v3/transport/transport.go b/cdc/scheduler/internal/v3/transport/transport.go index ade03146df..8fb2d86754 100644 --- a/cdc/scheduler/internal/v3/transport/transport.go +++ b/cdc/scheduler/internal/v3/transport/transport.go @@ -159,13 +159,6 @@ func (t *p2pTransport) Send( return errors.Trace(err) } } - - if len(msgs) != 0 { - log.Debug("schedulerv3: all messages sent", - zap.String("namespace", t.changefeed.Namespace), - zap.String("changefeed", t.changefeed.ID), - zap.Int("len", len(msgs))) - } return nil } diff --git a/cdc/sink/dmlsink/cloudstorage/encoding_worker.go b/cdc/sink/dmlsink/cloudstorage/encoding_worker.go index 0ba69ef395..55dcdd9d25 100644 --- a/cdc/sink/dmlsink/cloudstorage/encoding_worker.go +++ b/cdc/sink/dmlsink/cloudstorage/encoding_worker.go @@ -17,10 +17,8 @@ import ( "sync/atomic" "github.com/pingcap/errors" - "github.com/pingcap/log" "github.com/pingcap/tiflow/cdc/model" "github.com/pingcap/tiflow/pkg/sink/codec" - "go.uber.org/zap" "golang.org/x/sync/errgroup" ) @@ -52,10 +50,6 @@ func newEncodingWorker( } func (w *encodingWorker) run(ctx context.Context) error { - log.Debug("encoding worker started", zap.Int("workerID", w.id), - zap.String("namespace", w.changeFeedID.Namespace), - zap.String("changefeed", w.changeFeedID.ID)) - eg, ctx := errgroup.WithContext(ctx) eg.Go(func() error { for { diff --git a/pkg/config/sink.go b/pkg/config/sink.go index ff8c1236ff..9b2f77b601 100644 --- a/pkg/config/sink.go +++ b/pkg/config/sink.go @@ -238,7 +238,6 @@ func (s *SinkConfig) ShouldSendAllBootstrapAtStart() bool { return false } should := s.ShouldSendBootstrapMsg() && util.GetOrZero(s.SendAllBootstrapAtStart) - log.Info("should send all bootstrap at start", zap.Bool("should", should)) return should } @@ -845,10 +844,6 @@ func (s *SinkConfig) validateAndAdjustSinkURI(sinkURI *url.URL) error { return err } - log.Info("succeed to parse parameter from sink uri", - zap.String("protocol", util.GetOrZero(s.Protocol)), - zap.String("txnAtomicity", string(util.GetOrZero(s.TxnAtomicity)))) - // Check that protocol config is compatible with the scheme. if sink.IsMySQLCompatibleScheme(sinkURI.Scheme) && s.Protocol != nil { return cerror.ErrSinkURIInvalid.GenWithStackByArgs(fmt.Sprintf("protocol %s "+ diff --git a/pkg/txnutil/lock_resolver.go b/pkg/txnutil/lock_resolver.go index 6184dc0045..7dad490c6d 100644 --- a/pkg/txnutil/lock_resolver.go +++ b/pkg/txnutil/lock_resolver.go @@ -16,17 +16,14 @@ package txnutil import ( "bytes" "context" - "time" "github.com/pingcap/errors" "github.com/pingcap/kvproto/pkg/kvrpcpb" - "github.com/pingcap/log" "github.com/pingcap/tiflow/cdc/model" tikverr "github.com/tikv/client-go/v2/error" "github.com/tikv/client-go/v2/tikv" "github.com/tikv/client-go/v2/tikvrpc" "github.com/tikv/client-go/v2/txnkv" - "go.uber.org/zap" ) // LockResolver resolves lock in the given region. @@ -54,24 +51,6 @@ const scanLockLimit = 1024 func (r *resolver) Resolve(ctx context.Context, regionID uint64, maxVersion uint64) (err error) { var totalLocks []*txnkv.Lock - start := time.Now() - - defer func() { - // Only log when there are locks or error to avoid log flooding. - if len(totalLocks) != 0 || err != nil { - cost := time.Since(start) - log.Info("resolve lock finishes", - zap.Uint64("regionID", regionID), - zap.Int("lockCount", len(totalLocks)), - zap.Any("locks", totalLocks), - zap.Uint64("maxVersion", maxVersion), - zap.String("namespace", r.changefeed.Namespace), - zap.String("changefeed", r.changefeed.ID), - zap.Duration("duration", cost), - zap.Error(err)) - } - }() - // TODO test whether this function will kill active transaction req := tikvrpc.NewRequest(tikvrpc.CmdScanLock, &kvrpcpb.ScanLockRequest{ MaxVersion: maxVersion, From 71768624e3b44b17f62d61f4ca106e8d5b2d6f79 Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Tue, 14 Apr 2026 17:13:24 +0800 Subject: [PATCH 2/4] remove some verbose and useless logs --- pkg/txnutil/lock_resolver.go | 3 --- 1 file changed, 3 deletions(-) diff --git a/pkg/txnutil/lock_resolver.go b/pkg/txnutil/lock_resolver.go index 7dad490c6d..26b5cfab75 100644 --- a/pkg/txnutil/lock_resolver.go +++ b/pkg/txnutil/lock_resolver.go @@ -49,8 +49,6 @@ func NewLockerResolver( const scanLockLimit = 1024 func (r *resolver) Resolve(ctx context.Context, regionID uint64, maxVersion uint64) (err error) { - var totalLocks []*txnkv.Lock - // TODO test whether this function will kill active transaction req := tikvrpc.NewRequest(tikvrpc.CmdScanLock, &kvrpcpb.ScanLockRequest{ MaxVersion: maxVersion, @@ -109,7 +107,6 @@ func (r *resolver) Resolve(ctx context.Context, regionID uint64, maxVersion uint for i := range locksInfo { locks[i] = txnkv.NewLock(locksInfo[i]) } - totalLocks = append(totalLocks, locks...) _, err1 := r.kvStorage.GetLockResolver().ResolveLocks(bo, 0, locks) if err1 != nil { From 30338d27dda482426a824b41735a9030a4115dcf Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Thu, 16 Apr 2026 11:16:23 +0800 Subject: [PATCH 3/4] bump sarama to avoid verbose logs --- go.mod | 2 +- go.sum | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/go.mod b/go.mod index ec9f8a7ce4..4929bce08d 100644 --- a/go.mod +++ b/go.mod @@ -439,7 +439,7 @@ require ( sourcegraph.com/sourcegraph/appdash-data v0.0.0-20151005221446-73f23eafcf67 // indirect ) -replace github.com/IBM/sarama v1.41.2 => github.com/pingcap/sarama v1.41.2-pingcap-20250415 +replace github.com/IBM/sarama v1.41.2 => github.com/pingcap/sarama v1.41.2-pingcap-20251202-x // Fix https://github.com/pingcap/tiflow/issues/4961 replace github.com/benbjohnson/clock v1.3.5 => github.com/benbjohnson/clock v1.1.0 diff --git a/go.sum b/go.sum index 1cabd51f1c..d7c28da508 100644 --- a/go.sum +++ b/go.sum @@ -888,8 +888,8 @@ github.com/pingcap/log v1.1.1-0.20250917021125-19901e015dc9 h1:qG9BSvlWFEE5otQGa github.com/pingcap/log v1.1.1-0.20250917021125-19901e015dc9/go.mod h1:ORfBOFp1eteu2odzsyaxI+b8TzJwgjwyQcGhI+9SfEA= github.com/pingcap/metering_sdk v0.0.0-20251110022152-dac449ac5389 h1:bqbE3bwFSrUDSiN5M4EG+IXmm5eWLJnGRy/caXnxuHA= github.com/pingcap/metering_sdk v0.0.0-20251110022152-dac449ac5389/go.mod h1:zie1N5PRttgtqkZmRtpIDM7CuyWtvlX9LTxRd3fVSc4= -github.com/pingcap/sarama v1.41.2-pingcap-20250415 h1:jc/31lgAuSWLh8zr3y5bL0atpBFHAjch5H1Nnlb04J0= -github.com/pingcap/sarama v1.41.2-pingcap-20250415/go.mod h1:Kwi9CT6CuDYad3KS7HqjsbmD2DWkIKI7qI6a8PKlGb4= +github.com/pingcap/sarama v1.41.2-pingcap-20251202-x h1:9Vi3qqyDNZxG6fnXQhpeTsnwzSBWNpMeb8o02JkL9JM= +github.com/pingcap/sarama v1.41.2-pingcap-20251202-x/go.mod h1:xdpu7sd6OE1uxNdjYTSKUfY8FaKkJES9/+EyjSgiGQk= github.com/pingcap/sysutil v1.0.1-0.20240311050922-ae81ee01f3a5 h1:T4pXRhBflzDeAhmOQHNPRRogMYxP13V7BkYw3ZsoSfE= github.com/pingcap/sysutil v1.0.1-0.20240311050922-ae81ee01f3a5/go.mod h1:rlimy0GcTvjiJqvD5mXTRr8O2eNZPBrcUgiWVYp9530= github.com/pingcap/tidb v1.1.0-beta.0.20260325043212-0c4df2e19ecc h1:2Ateg1PUQqozi+TtbL+Bx62qn4IpQ48VTOpqCh2d3AI= From 8c02fc9896bdbf253cc75bde3c5d34f39a7ebbdd Mon Sep 17 00:00:00 2001 From: 3AceShowHand Date: Thu, 16 Apr 2026 11:24:31 +0800 Subject: [PATCH 4/4] update a lot of logs --- cdc/capture/capture.go | 8 ++++++-- cdc/model/changefeed.go | 6 +----- cdc/model/sink.go | 2 +- pkg/migrate/migrate.go | 2 -- pkg/sink/pulsar/factory.go | 1 - pkg/txnutil/gc/gc_manager.go | 3 --- 6 files changed, 8 insertions(+), 14 deletions(-) diff --git a/cdc/capture/capture.go b/cdc/capture/capture.go index 7b46150b19..dbb3c3e37b 100644 --- a/cdc/capture/capture.go +++ b/cdc/capture/capture.go @@ -224,7 +224,7 @@ func (c *captureImpl) reset(ctx context.Context) (*vars.GlobalVars, error) { if err != nil { return nil, errors.Trace(err) } - log.Info("reset session successfully", zap.Any("session", sess)) + log.Info("reset session successfully", zap.Int64("leaseID", int64(lease.ID))) c.captureMu.Lock() defer c.captureMu.Unlock() @@ -291,7 +291,11 @@ func (c *captureImpl) reset(ctx context.Context) (*vars.GlobalVars, error) { c.processorManager = c.newProcessorManager( c.info, c.upstreamManager, &c.liveness, c.config.Debug.Scheduler, globalVars) - log.Info("capture initialized", zap.Any("capture", c.info)) + log.Info("capture initialized", + zap.String("captureID", c.info.ID), + zap.String("advertiseAddr", c.info.AdvertiseAddr), + zap.String("version", c.info.Version), + zap.String("gitHash", c.info.GitHash)) return globalVars, nil } diff --git a/cdc/model/changefeed.go b/cdc/model/changefeed.go index 1f009d11f7..be23f64c78 100644 --- a/cdc/model/changefeed.go +++ b/cdc/model/changefeed.go @@ -218,8 +218,7 @@ func (info *ChangeFeedInfo) NeedBlockGC() bool { func (info *ChangeFeedInfo) isFailedByGC() bool { if info.Error == nil { - log.Panic("changefeed info is not consistent", - zap.Any("state", info.State), zap.Any("error", info.Error)) + log.Panic("changefeed info is not consistent", zap.Any("state", info.State)) } return cerror.IsChangefeedGCFastFailErrorCode(errors.RFCErrorCode(info.Error.Code)) } @@ -381,9 +380,6 @@ func (info *ChangeFeedInfo) RmUnusedFields() { } func (info *ChangeFeedInfo) rmMQOnlyFields() { - log.Info("since the downstream is not a MQ, remove MQ only fields", - zap.String("namespace", info.Namespace), - zap.String("changefeed", info.ID)) info.Config.Sink.DispatchRules = nil info.Config.Sink.SchemaRegistry = nil info.Config.Sink.EncoderConcurrency = nil diff --git a/cdc/model/sink.go b/cdc/model/sink.go index 54ccf98d3e..fb60960451 100644 --- a/cdc/model/sink.go +++ b/cdc/model/sink.go @@ -1193,7 +1193,7 @@ func trySplitAndSortUpdateEvent( split := false for _, e := range events { if e == nil { - log.Warn("skip emit nil event", zap.Any("event", e)) + log.Warn("skip emit nil event") continue } diff --git a/pkg/migrate/migrate.go b/pkg/migrate/migrate.go index d7723a15ab..7b64ae015c 100644 --- a/pkg/migrate/migrate.go +++ b/pkg/migrate/migrate.go @@ -279,9 +279,7 @@ func (m *migrator) migrate(ctx context.Context, etcdNoMetaVersion bool, oldVersi log.Error("update meta version failed, etcd meta data migration failed", zap.Error(err)) return cerror.WrapError(cerror.ErrEtcdMigrateFailed, err) } - log.Info("etcd data migration successful") cleanOldData(ctx, m.cli.GetEtcdClient()) - log.Info("clean old etcd data successful") return nil } diff --git a/pkg/sink/pulsar/factory.go b/pkg/sink/pulsar/factory.go index abd1a81cb1..d71fdcf133 100644 --- a/pkg/sink/pulsar/factory.go +++ b/pkg/sink/pulsar/factory.go @@ -131,6 +131,5 @@ func setupAuthentication(config *config.PulsarConfig) (bool, pulsar.Authenticati func NewMockCreatorFactory(config *config.PulsarConfig, changefeedID model.ChangeFeedID, sinkConfig *config.SinkConfig, ) (pulsar.Client, error) { - log.Info("mock pulsar client factory created", zap.Any("changfeedID", changefeedID)) return nil, nil } diff --git a/pkg/txnutil/gc/gc_manager.go b/pkg/txnutil/gc/gc_manager.go index 64fa462e35..a7aafeb500 100644 --- a/pkg/txnutil/gc/gc_manager.go +++ b/pkg/txnutil/gc/gc_manager.go @@ -90,9 +90,6 @@ func (m *gcManager) TryUpdateGCSafePoint( failpoint.Inject("InjectActualGCSafePoint", func(val failpoint.Value) { actual = uint64(val.(int)) }) - if actual == checkpointTs { - log.Info("update gc safe point success", zap.Uint64("gcSafePointTs", checkpointTs)) - } if actual > checkpointTs { log.Warn("update gc safe point failed, the gc safe point is larger than checkpointTs", zap.Uint64("actual", actual), zap.Uint64("checkpointTs", checkpointTs))