redo(ticdc): enable pprof and set memory limit for redo applier (ping…

…cap#10904) (pingcap#10995)

close pingcap#10900

cdc/redo/reader/reader.go

@@ -37,7 +37,7 @@ import (
 const (
 	emitBatch             = mysql.DefaultMaxTxnRow
 	defaultReaderChanSize = mysql.DefaultWorkerCount * emitBatch
-	maxTotalMemoryUsage   = 90.0
+	maxTotalMemoryUsage   = 80.0
 	maxWaitDuration       = time.Minute * 2
 )
 
@@ -221,11 +221,6 @@ func (l *LogReader) runReader(egCtx context.Context, cfg *readerConfig) error {
 				case l.rowCh <- row:
 				}
 			}
-			err := util.WaitMemoryAvailable(maxTotalMemoryUsage, maxWaitDuration)
-			if err != nil {
-				return errors.Trace(err)
-			}
-
 		case redo.RedoDDLLogFileType:
 			ddl := model.LogToDDL(item.data.RedoDDL)
 			if ddl != nil && ddl.CommitTs > cfg.startTs && ddl.CommitTs <= cfg.endTs {

pkg/cmd/redo/apply.go

@@ -14,18 +14,27 @@
 package redo
 
 import (
+	"net/http"
+	_ "net/http/pprof" //nolint:gosec
 	"net/url"
+	"runtime/debug"
+	"time"
 
+	"github.com/pingcap/log"
 	"github.com/pingcap/tiflow/pkg/applier"
 	cmdcontext "github.com/pingcap/tiflow/pkg/cmd/context"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
+	"github.com/pingcap/tiflow/pkg/util"
 	"github.com/spf13/cobra"
+	"go.uber.org/zap"
 )
 
 // applyRedoOptions defines flags for the `redo apply` command.
 type applyRedoOptions struct {
 	options
-	sinkURI string
+	sinkURI              string
+	enableProfiling      bool
+	memoryLimitInGiBytes int64
 }
 
 // newapplyRedoOptions creates new applyRedoOptions for the `redo apply` command.
@@ -39,6 +48,8 @@ func (o *applyRedoOptions) addFlags(cmd *cobra.Command) {
 	cmd.Flags().StringVar(&o.sinkURI, "sink-uri", "", "target database sink-uri")
 	// the possible error returned from MarkFlagRequired is `no such flag`
 	cmd.MarkFlagRequired("sink-uri") //nolint:errcheck
+	cmd.Flags().BoolVar(&o.enableProfiling, "enable-profiling", true, "enable pprof profiling")
+	cmd.Flags().Int64Var(&o.memoryLimitInGiBytes, "memory-limit", 10, "memory limit in GiB")
 }
 
 //nolint:unparam
@@ -55,13 +66,38 @@ func (o *applyRedoOptions) complete(cmd *cobra.Command) error {
 		sinkURI.RawQuery = rawQuery.Encode()
 		o.sinkURI = sinkURI.String()
 	}
+
+	totalMemory, err := util.GetMemoryLimit()
+	if err == nil {
+		totalMemoryInBytes := int64(float64(totalMemory) * 0.8)
+		memoryLimitInBytes := o.memoryLimitInGiBytes * 1024 * 1024 * 1024
+		if totalMemoryInBytes != 0 && memoryLimitInBytes > totalMemoryInBytes {
+			memoryLimitInBytes = totalMemoryInBytes
+		}
+		debug.SetMemoryLimit(memoryLimitInBytes)
+		log.Info("set memory limit", zap.Int64("memoryLimit", memoryLimitInBytes))
+	}
+
 	return nil
 }
 
 // run runs the `redo apply` command.
 func (o *applyRedoOptions) run(cmd *cobra.Command) error {
 	ctx := cmdcontext.GetDefaultContext()
 
+	if o.enableProfiling {
+		go func() {
+			server := &http.Server{
+				Addr:              ":6060",
+				ReadHeaderTimeout: 5 * time.Second,
+			}
+			log.Info("Start http pprof server", zap.String("addr", server.Addr))
+			if err := server.ListenAndServe(); err != nil {
+				log.Fatal("http pprof", zap.Error(err))
+			}
+		}()
+	}
+
 	cfg := &applier.RedoApplierConfig{
 		Storage: o.storage,
 		SinkURI: o.sinkURI,