[codex] Fix Redis long-run memory retention

.github/workflows/redis_long_run.yml

@@ -0,0 +1,80 @@
+name: Redis Long Run
+
+on:
+  workflow_dispatch:
+    inputs:
+      rounds:
+        description: "Number of Redis traffic rounds after warmup"
+        required: false
+        default: "8"
+      batch_size:
+        description: "Redis commands per round"
+        required: false
+        default: "2000"
+      growth_limit_mb:
+        description: "Allowed heap growth after warmup in MB"
+        required: false
+        default: "64"
+
+permissions:
+  contents: read
+
+jobs:
+  redis-long-run:
+    runs-on: ubuntu-22.04
+    timeout-minutes: 90
+    steps:
+      - uses: actions/checkout@v4.2.2
+        with:
+          submodules: recursive
+
+      - name: Set up Go
+        uses: actions/setup-go@v5.3.0
+        with:
+          go-version: '1.23.3'
+
+      - name: Build
+        run: |
+          /bin/bash init_env.sh
+          make clean && make build-bpf && make
+
+      - name: Provision LVH VM
+        uses: cilium/little-vm-helper@v0.0.28
+        with:
+          test-name: redis-long-run
+          image-version: '5.15-20240912.022020'
+          cpu: 2
+          mem: '4G'
+          host-mount: ./
+          install-dependencies: 'true'
+          cmd: |
+            chmod +x /host/kyanos
+
+      - name: Install VM dependencies
+        uses: cilium/little-vm-helper@v0.0.28
+        with:
+          provision: 'false'
+          cmd: |
+            apt-get update
+            apt-get install -y redis-tools python3 curl
+
+      - name: Run Redis long-run test
+        uses: cilium/little-vm-helper@v0.0.28
+        env:
+          REDIS_LONG_RUN_ROUNDS: ${{ github.event.inputs.rounds || '8' }}
+          REDIS_LONG_RUN_BATCH_SIZE: ${{ github.event.inputs.batch_size || '2000' }}
+          REDIS_LONG_RUN_GROWTH_LIMIT_MB: ${{ github.event.inputs.growth_limit_mb || '64' }}
+          REDIS_LONG_RUN_OUTPUT_DIR: /host/testdata/output/redis-long-run
+        with:
+          provision: 'false'
+          cmd: |
+            set -euxo pipefail
+            mkdir -p /host/testdata/output/redis-long-run
+            bash /host/testdata/test_redis_long_run.sh 'sudo /host/kyanos'
+
+      - name: Upload long-run artifacts
+        if: always()
+        uses: actions/upload-artifact@v4.6.1
+        with:
+          name: redis-long-run-artifacts
+          path: testdata/output/redis-long-run

agent/agent.go

@@ -17,6 +17,7 @@ import (
 	"kyanos/bpf/loader"
 	"kyanos/common"
 	"kyanos/version"
+	"net/http"
 	"os"
 	"os/exec"
 	"os/signal"
@@ -26,7 +27,7 @@ import (
 	"syscall"
 	"time"
 
-	_ "net/http/pprof"
+	"net/http/pprof"
 
 	"github.com/cilium/ebpf/rlimit"
 	gops "github.com/google/gops/agent"
@@ -70,6 +71,7 @@ func SetupAgent(options ac.AgentOptions) {
 		context.Background(), syscall.SIGINT, syscall.SIGTERM,
 	)
 	options.Ctx = ctx
+	startPprofServer(options)
 
 	defer stopFunc()
 
@@ -252,3 +254,49 @@ func startGopsServer(opts ac.AgentOptions) {
 		}
 	}
 }
+
+func startPprofServer(opts ac.AgentOptions) {
+	if !opts.EnablePprof {
+		return
+	}
+
+	addr := opts.GetPprofAddr()
+	displayAddr := addr
+	if strings.HasPrefix(addr, ":") {
+		displayAddr = "localhost" + addr
+	}
+
+	mux := http.NewServeMux()
+	mux.HandleFunc("/debug/pprof/", pprof.Index)
+	mux.HandleFunc("/debug/pprof/cmdline", pprof.Cmdline)
+	mux.HandleFunc("/debug/pprof/profile", pprof.Profile)
+	mux.HandleFunc("/debug/pprof/symbol", pprof.Symbol)
+	mux.HandleFunc("/debug/pprof/trace", pprof.Trace)
+	mux.HandleFunc("/debug/pprof/{name}", pprof.Index)
+
+	server := &http.Server{
+		Addr:              addr,
+		Handler:           mux,
+		ReadHeaderTimeout: 10 * time.Second,
+		IdleTimeout:       120 * time.Second,
+	}
+
+	go func() {
+		common.AgentLog.Infof("Starting pprof server on http://%s/debug/pprof/", displayAddr)
+		common.AgentLog.Info("pprof endpoints: /debug/pprof/heap, /debug/pprof/profile, /debug/pprof/goroutine")
+		if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed {
+			common.AgentLog.Errorf("Failed to start pprof server: %v", err)
+		}
+	}()
+
+	go func() {
+		<-opts.Ctx.Done()
+		shutdownCtx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+		defer cancel()
+		if err := server.Shutdown(shutdownCtx); err != nil {
+			common.AgentLog.Errorf("Failed to shutdown pprof server: %v", err)
+		} else {
+			common.AgentLog.Info("pprof server shutdown gracefully")
+		}
+	}()
+}

agent/common/options.go

@@ -50,6 +50,8 @@ type AgentOptions struct {
 	ConntrackCloseWaitTimeMills int
 	MaxAllowStuckTimeMills      int
 	StartGopsServer             bool
+	EnablePprof                 bool
+	PprofAddr                   string
 
 	FilterComm              string
 	ProcessExecEventChannel chan *bpf.AgentProcessExecEvent
@@ -74,6 +76,13 @@ type AgentOptions struct {
 	FirstPacketEventMapPageNum int
 }
 
+func (o AgentOptions) GetPprofAddr() string {
+	if o.PprofAddr == "" {
+		return "localhost:6060"
+	}
+	return o.PprofAddr
+}
+
 func (o AgentOptions) FilterByContainer() bool {
 	return o.ContainerId != "" || o.ContainerName != "" || o.PodName != ""
 }

agent/protocol/common.go

@@ -4,6 +4,11 @@ import (
 	"kyanos/agent/buffer"
 )
 
+const (
+	maxPendingParsedMessages      = 1024
+	maxPendingParsedMessagesBytes = 16 * 1024 * 1024
+)
+
 func matchByTimestamp(reqStream *ParsedMessageQueue, respStream *ParsedMessageQueue) []Record {
 	if len(*reqStream) == 0 || len(*respStream) == 0 {
 		return nil
@@ -36,6 +41,41 @@ func matchByTimestamp(reqStream *ParsedMessageQueue, respStream *ParsedMessageQu
 	return records
 }
 
+func trimPendingParsedMessages(queue *ParsedMessageQueue, maxCount int, maxBytes int) {
+	if queue == nil || len(*queue) == 0 {
+		return
+	}
+
+	start := len(*queue)
+	keptCount := 0
+	keptBytes := 0
+	for i := len(*queue) - 1; i >= 0; i-- {
+		msgBytes := max(1, (*queue)[i].ByteSize())
+		if keptCount > 0 && (keptCount+1 > maxCount || keptBytes+msgBytes > maxBytes) {
+			break
+		}
+		start = i
+		keptCount++
+		keptBytes += msgBytes
+	}
+
+	if start > 0 {
+		*queue = (*queue)[start:]
+	}
+}
+
+func trimPendingParsedMessagesForStream(streams map[StreamId]*ParsedMessageQueue, streamID StreamId) {
+	queue, ok := streams[streamID]
+	if !ok || queue == nil {
+		return
+	}
+
+	trimPendingParsedMessages(queue, maxPendingParsedMessages, maxPendingParsedMessagesBytes)
+	if len(*queue) == 0 {
+		delete(streams, streamID)
+	}
+}
+
 func CreateFrameBase(streamBuffer *buffer.StreamBuffer, readBytes int) (FrameBase, bool) {
 	seq := streamBuffer.Head().LeftBoundary()
 	ts, ok := streamBuffer.FindTimestampBySeq(seq)

agent/protocol/redis..go

@@ -382,9 +382,14 @@ func (r *RedisStreamParser) Match(reqStreams map[StreamId]*ParsedMessageQueue, r
 	reqStream, ok1 := reqStreams[0]
 	respStream, ok2 := respStreams[0]
 	if !ok1 || !ok2 {
+		trimPendingParsedMessagesForStream(reqStreams, 0)
+		trimPendingParsedMessagesForStream(respStreams, 0)
 		return []Record{}
 	}
-	return matchByTimestamp(reqStream, respStream)
+	records := matchByTimestamp(reqStream, respStream)
+	trimPendingParsedMessagesForStream(reqStreams, 0)
+	trimPendingParsedMessagesForStream(respStreams, 0)
+	return records
 }
 func ParseSize(decoder *BinaryDecoder) (int, error) {
 	str, err := decoder.ExtractStringUntil(kTerminalSequence)

agent/protocol/redis_test.go

@@ -0,0 +1,71 @@
+package protocol
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+func newTestRedisMessage(seq uint64, timestamp uint64, byteSize int, isReq bool) *RedisMessage {
+	return &RedisMessage{
+		FrameBase: NewFrameBase(timestamp, byteSize, seq),
+		payload:   fmt.Sprintf("payload-%d", seq),
+		command:   "GET",
+		isReq:     isReq,
+	}
+}
+
+func TestRedisMatchTrimsPendingRequestsWithoutResponses(t *testing.T) {
+	reqQueue := ParsedMessageQueue{}
+	for i := 0; i < maxPendingParsedMessages+128; i++ {
+		reqQueue = append(reqQueue, newTestRedisMessage(uint64(i+1), uint64(i+1), 64, true))
+	}
+
+	parser := RedisStreamParser{}
+	reqStreams := map[StreamId]*ParsedMessageQueue{0: &reqQueue}
+
+	records := parser.Match(reqStreams, map[StreamId]*ParsedMessageQueue{})
+
+	require.Empty(t, records)
+	require.Contains(t, reqStreams, StreamId(0))
+	assert.Len(t, *reqStreams[0], maxPendingParsedMessages)
+	assert.Equal(t, uint64(129), (*reqStreams[0])[0].Seq())
+}
+
+func TestRedisMatchTrimsPendingResponsesWithoutRequests(t *testing.T) {
+	respQueue := ParsedMessageQueue{}
+	for i := 0; i < maxPendingParsedMessages+64; i++ {
+		respQueue = append(respQueue, newTestRedisMessage(uint64(i+1), uint64(i+1), 64, false))
+	}
+
+	parser := RedisStreamParser{}
+	respStreams := map[StreamId]*ParsedMessageQueue{0: &respQueue}
+
+	records := parser.Match(map[StreamId]*ParsedMessageQueue{}, respStreams)
+
+	require.Empty(t, records)
+	require.Contains(t, respStreams, StreamId(0))
+	assert.Len(t, *respStreams[0], maxPendingParsedMessages)
+	assert.Equal(t, uint64(65), (*respStreams[0])[0].Seq())
+}
+
+func TestRedisMatchRemovesEmptyQueuesAfterSuccessfulMatch(t *testing.T) {
+	reqQueue := ParsedMessageQueue{
+		newTestRedisMessage(1, 100, 16, true),
+	}
+	respQueue := ParsedMessageQueue{
+		newTestRedisMessage(2, 200, 16, false),
+	}
+
+	parser := RedisStreamParser{}
+	reqStreams := map[StreamId]*ParsedMessageQueue{0: &reqQueue}
+	respStreams := map[StreamId]*ParsedMessageQueue{0: &respQueue}
+
+	records := parser.Match(reqStreams, respStreams)
+
+	require.Len(t, records, 1)
+	assert.NotContains(t, reqStreams, StreamId(0))
+	assert.NotContains(t, respStreams, StreamId(0))
+}

cmd/root.go

@@ -105,6 +105,8 @@ func init() {
 	rootCmd.PersistentFlags().IntVar(&options.MaxAllowStuckTimeMills, "max-allow-stuck-time-mills", 1000, "--max-allow-stuck-time-mills 100")
 
 	rootCmd.PersistentFlags().BoolVar(&options.StartGopsServer, "gops", false, "start gops server")
+	rootCmd.PersistentFlags().BoolVar(&options.EnablePprof, "pprof", false, "enable pprof HTTP endpoint for profiling (WARNING: exposes sensitive profiling data; do not use on untrusted networks)")
+	rootCmd.PersistentFlags().StringVar(&options.PprofAddr, "pprof-addr", "127.0.0.1:6060", "pprof HTTP server address (WARNING: binding to non-loopback addresses, e.g. \":6060\" or \"0.0.0.0:6060\", will expose profiling data to the network)")
 
 	rootCmd.PersistentFlags().MarkHidden("default-log-level")
 	rootCmd.PersistentFlags().MarkHidden("agent-log-level")

docs/cn/debug-tips.md

@@ -85,6 +85,50 @@ VSCODE 直接打开项目即可，.vscode/launch.json 添加配置如下：
 
 注意添加 `--debug-output` 参数。
 
+## 性能分析（Profiling）
+
+Kyanos 提供了 pprof 端点用于运行时性能分析和调试。你可以使用 `--pprof` 标志启用 pprof HTTP 服务器：
+
+```bash
+./kyanos watch --pprof
+```
+
+默认情况下，pprof 服务器监听 `localhost:6060`。你可以使用 `--pprof-addr` 标志自定义监听地址：
+
+```bash
+./kyanos watch --pprof --pprof-addr="0.0.0.0:9090"
+```
+
+可用的 pprof 端点：
+
+| 端点 | 说明 |
+|------|------|
+| `/debug/pprof/` | 索引页面，显示所有可用的 profile |
+| `/debug/pprof/heap` | 内存堆 profile |
+| `/debug/pprof/profile` | CPU profile（默认 30 秒） |
+| `/debug/pprof/goroutine` | Goroutine 堆栈信息 |
+| `/debug/pprof/allocs` | 内存分配 profile |
+| `/debug/pprof/block` | 阻塞 profile |
+| `/debug/pprof/mutex` | 锁竞争 profile |
+
+使用示例：
+
+```bash
+# 采集 CPU profile
+curl -o cpu.pprof http://localhost:6060/debug/pprof/profile?seconds=30
+go tool pprof cpu.pprof
+
+# 查看堆内存 profile
+go tool pprof http://localhost:6060/debug/pprof/heap
+
+# 查看 goroutine 堆栈
+curl http://localhost:6060/debug/pprof/goroutine?debug=1
+```
+
+> [!TIP]
+>
+> pprof 端点对于诊断 Kyanos 自身的性能问题、内存泄漏或 goroutine 泄漏非常有用。
+
 ## 源码结构
 
 ```