feat: add stream consumer group backlog report

README.md

@@ -19,6 +19,7 @@ It provides abilities to:
 - Convert RDB files to Redis Serialization Protocol (or AOF file)
 - Find the biggest N keys in RDB files
 - Find the hottest N keys by LFU frequency in RDB files
+- Report the pending backlog of stream consumer groups
 - Draw FlameGraph to analysis which kind of keys occupied most memory
 - Customize data usage
 - Generate RDB file
@@ -54,7 +55,7 @@ use `rdb` command in terminal, you can see it's manual
 $ rdb
 This is a tool to parse Redis' RDB files
 Options:
-  -c command, including: json/memory/aof/bigkey/hotkey/prefix/flamegraph
+  -c command, including: json/memory/aof/bigkey/hotkey/prefix/flamegraph/streambacklog
   -o output file path
   -n number of result, using in command: bigkey/hotkey/prefix
   -port listen port for flame graph web service
@@ -96,6 +97,8 @@ parameters between '[' and ']' is optional
   rdb -c flamegraph [-port 16379] [-sep :] dump.rdb
 7. get hottest keys by LFU frequency (requires maxmemory-policy allkeys-lfu/volatile-lfu)
   rdb -c hotkey [-o hotkey.csv] [-n 50] dump.rdb
+8. report the pending backlog of every stream consumer group
+  rdb -c streambacklog [-o backlog.csv] dump.rdb
 ```
 
 # Convert to Json
@@ -561,6 +564,33 @@ database,key,type,size,size_readable,freq
 
 Note: Keys without LFU information are skipped. If the RDB file was not generated under LFU eviction policy, the output will be empty.
 
+# Stream Consumer Group Backlog
+
+For every consumer group of every stream, RDB can report how far the group is behind:
+
+```
+rdb -c streambacklog [-o <output_path>] <source_path>
+```
+
+Example:
+
+```
+rdb -c streambacklog cases/stream_listpacks_1.rdb
+```
+
+The CSV has one row for each consumer that holds pending entries, ordered by group name, then by descending `pending`.
+Pending entries that no consumer holds are collected into one row with an empty `consumer`, and a group without pending entries still gets a single row.
+`pending` is the number of entries the consumer holds, `oldest_delivery_time` and `max_delivery_count` are the smallest delivery time and the largest delivery count among them.
+`undelivered` is the number of messages the group has never been delivered, and `dangling` is the number of its pending entries whose message was deleted or is no longer in the stream.
+The last two columns describe the whole group, so every row of a group carries the same values.
+
+```csv
+database,key,group,consumer,pending,oldest_delivery_time,max_delivery_count,undelivered,dangling
+0,listpack,g1,c1,2,1528516636879,1,144,0
+0,listpack,g1,c2,2,1528516649782,1,144,0
+0,listpack,g4,,0,0,0,0,0
+```
+
 # Convert to AOF
 
 Usage:

README_CN.md

@@ -40,7 +40,7 @@ go install github.com/hdt3213/rdb@latest
 $ rdb
 This is a tool to parse Redis' RDB files
 Options:
-  -c command, including: json/memory/aof/bigkey/hotkey/prefix/flamegraph
+  -c command, including: json/memory/aof/bigkey/hotkey/prefix/flamegraph/streambacklog
   -o output file path
   -n number of result, using in command: bigkey/hotkey/prefix
   -port listen port for flame graph web service
@@ -82,6 +82,8 @@ parameters between '[' and ']' is optional
   rdb -c flamegraph [-port 16379] [-sep :] dump.rdb
 7. get hottest keys by LFU frequency (requires maxmemory-policy allkeys-lfu/volatile-lfu)
   rdb -c hotkey [-o hotkey.csv] [-n 50] dump.rdb
+8. report the pending backlog of every stream consumer group
+  rdb -c streambacklog [-o backlog.csv] dump.rdb
 ```
 
 # 转换为 JSON 格式

cmd.go

@@ -12,7 +12,7 @@ import (
 const help = `
 This is a tool to parse Redis' RDB files
 Options:
-  -c command, including: json/memory/aof/bigkey/hotkey/prefix/flamegraph
+  -c command, including: json/memory/aof/bigkey/hotkey/prefix/flamegraph/streambacklog
   -o output file path
   -n number of result, using in command: bigkey/hotkey/prefix
   -port listen port for flame graph web service
@@ -54,6 +54,8 @@ parameters between '[' and ']' is optional
   rdb -c flamegraph [-port 16379] [-sep :] dump.rdb
 7. get hottest keys by LFU frequency (requires maxmemory-policy allkeys-lfu/volatile-lfu)
   rdb -c hotkey [-o hotkey.csv] [-n 50] dump.rdb
+8. report the pending backlog of every stream consumer group
+  rdb -c streambacklog [-o backlog.csv] dump.rdb
 `
 
 type separators []string
@@ -158,6 +160,8 @@ func main() {
 		} else {
 			err = helper.PrefixAnalyse(src, n, maxDepth, outputFile, options...)
 		}
+	case "streambacklog":
+		err = helper.StreamBacklogReport(src, outputFile, options...)
 	case "flamegraph":
 		_, err = helper.FlameGraph(src, port, seps, options...)
 		if err != nil {

helper/streambacklog.go

@@ -0,0 +1,71 @@
+package helper
+
+import (
+	"encoding/csv"
+	"errors"
+	"fmt"
+	"os"
+	"strconv"
+
+	"github.com/hdt3213/rdb/core"
+	"github.com/hdt3213/rdb/model"
+)
+
+// StreamBacklogReport writes a CSV report of the pending backlog of every consumer group of every stream in the rdb file
+func StreamBacklogReport(rdbFilename string, output *os.File, options ...interface{}) error {
+	if rdbFilename == "" {
+		return errors.New("src file path is required")
+	}
+	rdbFile, err := os.Open(rdbFilename)
+	if err != nil {
+		return fmt.Errorf("open rdb %s failed, %v", rdbFilename, err)
+	}
+	defer func() {
+		_ = rdbFile.Close()
+	}()
+	var dec decoder = core.NewDecoder(rdbFile)
+	if dec, err = wrapDecoder(dec, options...); err != nil {
+		return err
+	}
+	_, err = output.WriteString("database,key,group,consumer,pending,oldest_delivery_time,max_delivery_count,undelivered,dangling\n")
+	if err != nil {
+		return fmt.Errorf("write header failed: %v", err)
+	}
+	csvWriter := csv.NewWriter(output)
+	defer csvWriter.Flush()
+	var writeErr error
+	err = dec.Parse(func(object model.RedisObject) bool {
+		stream, ok := object.(*model.StreamObject)
+		if !ok {
+			return true
+		}
+		for _, row := range stream.ConsumerBacklog() {
+			writeErr = csvWriter.Write([]string{
+				strconv.Itoa(object.GetDBIndex()),
+				object.GetKey(),
+				row.Group,
+				row.Consumer,
+				strconv.Itoa(row.Pending),
+				strconv.FormatUint(row.OldestDeliveryTime, 10),
+				strconv.FormatUint(row.MaxDeliveryCount, 10),
+				strconv.Itoa(row.Undelivered),
+				strconv.Itoa(row.Dangling),
+			})
+			if writeErr != nil {
+				return false
+			}
+		}
+		return true
+	})
+	if writeErr != nil {
+		return fmt.Errorf("csv write failed: %v", writeErr)
+	}
+	if err != nil {
+		return err
+	}
+	csvWriter.Flush()
+	if err = csvWriter.Error(); err != nil {
+		return fmt.Errorf("csv write failed: %v", err)
+	}
+	return nil
+}

helper/streambacklog_test.go

@@ -0,0 +1,145 @@
+package helper
+
+import (
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"github.com/hdt3213/rdb/core"
+	"github.com/hdt3213/rdb/model"
+)
+
+const streamBacklogHeader = "database,key,group,consumer,pending,oldest_delivery_time,max_delivery_count,undelivered,dangling"
+
+func runStreamBacklogReport(t *testing.T, srcRdb string, options ...interface{}) []string {
+	t.Helper()
+	outputPath := filepath.Join(t.TempDir(), "backlog.csv")
+	output, err := os.Create(outputPath)
+	if err != nil {
+		t.Fatalf("create output file failed: %v", err)
+	}
+	err = StreamBacklogReport(srcRdb, output, options...)
+	if closeErr := output.Close(); closeErr != nil {
+		t.Fatalf("close output failed: %v", closeErr)
+	}
+	if err != nil {
+		t.Fatalf("StreamBacklogReport failed: %v", err)
+	}
+	content, err := os.ReadFile(outputPath)
+	if err != nil {
+		t.Fatalf("read output failed: %v", err)
+	}
+	return strings.Split(strings.TrimRight(string(content), "\n"), "\n")
+}
+
+func assertLines(t *testing.T, got, want []string) {
+	t.Helper()
+	if strings.Join(got, "\n") != strings.Join(want, "\n") {
+		t.Errorf("report =\n%s\nwant\n%s", strings.Join(got, "\n"), strings.Join(want, "\n"))
+	}
+}
+
+func TestStreamBacklogReport(t *testing.T) {
+	got := runStreamBacklogReport(t, filepath.Join("../cases", "stream_listoacks_3.rdb"))
+	assertLines(t, got, []string{
+		streamBacklogHeader,
+		"0,mystream,consumer-group-name,consumer-name,1,1704557998397,1,0,0",
+	})
+
+	got = runStreamBacklogReport(t, filepath.Join("../cases", "stream_listpacks_1.rdb"))
+	assertLines(t, got, []string{
+		streamBacklogHeader,
+		"0,listpack,g1,c1,2,1528516636879,1,144,0",
+		"0,listpack,g1,c2,2,1528516649782,1,144,0",
+		"0,listpack,g2,c1,1,1528516695691,1,83,0",
+		"0,listpack,g3,c1,2,1528516699993,1,81,0",
+		"0,listpack,g4,,0,0,0,0,0",
+	})
+
+	got = runStreamBacklogReport(t, filepath.Join("../cases", "stream_listpacks_2.rdb"))
+	assertLines(t, got, []string{streamBacklogHeader})
+}
+
+func TestStreamBacklogReportWithRegex(t *testing.T) {
+	srcRdb := filepath.Join("../cases", "stream_listpacks_1.rdb")
+	got := runStreamBacklogReport(t, srcRdb, WithRegexOption("^listpack$"))
+	if len(got) != 6 {
+		t.Errorf("got %d lines, want the header and five rows", len(got))
+	}
+	got = runStreamBacklogReport(t, srcRdb, WithRegexOption("^no-such-key$"))
+	assertLines(t, got, []string{streamBacklogHeader})
+}
+
+func TestStreamBacklogReportCsvQuotingAndOrder(t *testing.T) {
+	newStream := func(key, group string, consumers ...*model.StreamConsumer) *model.StreamObject {
+		return &model.StreamObject{
+			BaseObject: &model.BaseObject{Key: key},
+			Version:    3,
+			Length:     2,
+			LastId:     &model.StreamId{Ms: 2},
+			Entries: []*model.StreamEntry{{
+				FirstMsgId: &model.StreamId{Ms: 1},
+				Fields:     []string{"k"},
+				Msgs: []*model.StreamMessage{
+					{Id: &model.StreamId{Ms: 1}, Fields: map[string]string{"k": "v"}},
+					{Id: &model.StreamId{Ms: 2}, Fields: map[string]string{"k": "v"}},
+				},
+			}},
+			Groups: []*model.StreamGroup{{
+				Name:      group,
+				LastId:    &model.StreamId{Ms: 1},
+				Pending:   []*model.StreamNAck{{Id: &model.StreamId{Ms: 1}, DeliveryTime: 111, DeliveryCount: 2}},
+				Consumers: consumers,
+			}},
+		}
+	}
+	pending := []*model.StreamId{{Ms: 1}}
+	quoted := newStream("b,stream", "g,1",
+		&model.StreamConsumer{Name: `c"1`, Pending: pending},
+		&model.StreamConsumer{Name: "c2", Pending: pending},
+	)
+	plain := newStream("zeta", "zg", &model.StreamConsumer{Name: "zc", Pending: pending})
+
+	srcRdb := filepath.Join(t.TempDir(), "streams.rdb")
+	file, err := os.Create(srcRdb)
+	if err != nil {
+		t.Fatalf("create rdb failed: %v", err)
+	}
+	enc := core.NewEncoder(file)
+	steps := []func() error{
+		enc.WriteHeader,
+		func() error { return enc.WriteDBHeader(0, 2, 0) },
+		func() error { return enc.WriteStringObject("plain", []byte("v")) },
+		func() error { return enc.WriteStreamObject(quoted.Key, quoted) },
+		func() error { return enc.WriteDBHeader(3, 1, 0) },
+		func() error { return enc.WriteStreamObject(plain.Key, plain) },
+		enc.WriteEnd,
+	}
+	for i, step := range steps {
+		if err := step(); err != nil {
+			t.Fatalf("write rdb step %d failed: %v", i, err)
+		}
+	}
+	if err := file.Close(); err != nil {
+		t.Fatalf("close rdb failed: %v", err)
+	}
+
+	got := runStreamBacklogReport(t, srcRdb)
+	assertLines(t, got, []string{
+		streamBacklogHeader,
+		`0,"b,stream","g,1","c""1",1,111,2,1,0`,
+		`0,"b,stream","g,1",c2,1,111,2,1,0`,
+		`3,zeta,zg,zc,1,111,2,1,0`,
+	})
+}
+
+func TestStreamBacklogReportErrors(t *testing.T) {
+	err := StreamBacklogReport("", os.Stdout)
+	if err == nil || err.Error() != "src file path is required" {
+		t.Errorf("empty src: got %v", err)
+	}
+	if err = StreamBacklogReport(filepath.Join(t.TempDir(), "missing.rdb"), os.Stdout); err == nil {
+		t.Error("missing file: want an error")
+	}
+}