diff --git a/benchmark_runner/stat.go b/benchmark_runner/stat.go index f92154a..794a028 100644 --- a/benchmark_runner/stat.go +++ b/benchmark_runner/stat.go @@ -22,8 +22,8 @@ type CmdStat struct { latency uint64 // microseconds latency error bool timedOut bool - rx uint64 // bytes received - tx uint64 // bytes received + rx uint64 // bytes received (from Redis replies) + tx uint64 // bytes sent (request/command bytes) } func (c *CmdStat) StartTs() uint64 { diff --git a/benchmark_runner/stat_test.go b/benchmark_runner/stat_test.go new file mode 100644 index 0000000..2994a78 --- /dev/null +++ b/benchmark_runner/stat_test.go @@ -0,0 +1,65 @@ +package benchmark_runner + +import ( + "testing" + + hdrhistogram "github.com/HdrHistogram/hdrhistogram-go" +) + +// AddEntry's last two params are (rx, tx) in that order. Lock the mapping so a +// received-byte count lands in Rx() and a sent-byte count lands in Tx(); a +// regression here silently mislabels the TxBytes/RxBytes throughput metrics. +func TestAddEntryMapsRxTxInOrder(t *testing.T) { + const rxBytes = uint64(7) // bytes received (reply) + const txBytes = uint64(13) // bytes sent (request) + + s := NewStat().AddEntry([]byte("READ"), []byte("q1"), 1000, 42, false, false, rxBytes, txBytes) + + entries := s.CmdStats() + if len(entries) != 1 { + t.Fatalf("expected 1 entry, got %d", len(entries)) + } + if got := entries[0].Rx(); got != rxBytes { + t.Fatalf("Rx() = %d, want %d (received bytes)", got, rxBytes) + } + if got := entries[0].Tx(); got != txBytes { + t.Fatalf("Tx() = %d, want %d (sent bytes)", got, txBytes) + } +} + +func TestNewCmdStatMapsRxTxInOrder(t *testing.T) { + c := NewCmdStat([]byte("READ"), []byte("q1"), 42, false, false, 7, 13) + if c.Rx() != 7 { + t.Fatalf("Rx() = %d, want 7", c.Rx()) + } + if c.Tx() != 13 { + t.Fatalf("Tx() = %d, want 13", c.Tx()) + } +} + +// Guards the user-visible mapping in GetTotalsMap: the accumulated sent bytes +// (txTotalBytes) must surface under "TxBytes" and received under "RxBytes". +// This is the JSON the #111 symptom appeared in, and the aggregation label is +// otherwise untested. +func TestGetTotalsMapMapsTxRxCorrectly(t *testing.T) { + h := func() *hdrhistogram.Histogram { return hdrhistogram.New(1, 1_000_000_000, 3) } + b := &BenchmarkRunner{ + totalHistogram: h(), + setupWriteHistogram: h(), + writeHistogram: h(), + readHistogram: h(), + readCursorHistogram: h(), + updateHistogram: h(), + deleteHistogram: h(), + txTotalBytes: 13, // bytes sent + rxTotalBytes: 7, // bytes received + } + + configs := b.GetTotalsMap() + if got := configs["TxBytes"]; got != uint64(13) { + t.Fatalf("configs[\"TxBytes\"] = %v, want 13 (sent bytes)", got) + } + if got := configs["RxBytes"]; got != uint64(7) { + t.Fatalf("configs[\"RxBytes\"] = %v, want 7 (received bytes)", got) + } +} diff --git a/cmd/ftsb_redisearch/cmd_processor.go b/cmd/ftsb_redisearch/cmd_processor.go index d02e9aa..4336f95 100644 --- a/cmd/ftsb_redisearch/cmd_processor.go +++ b/cmd/ftsb_redisearch/cmd_processor.go @@ -302,7 +302,9 @@ func sendIfRequired(p *processor, client radix.Client, cmdType string, cmdQueryI took := uint64(duration.Microseconds()) rcv := replies[pos] rxBytesCount += getRxLen(rcv) - stat := benchmark_runner.NewStat().AddEntry([]byte(cmdType), []byte(cmdQueryId), uint64(t.Unix()), took, hadError, isTimeout, txBytesCount, rxBytesCount) + // AddEntry takes (..., rx, tx): received bytes first, then sent. + // txBytesCount is what we send to Redis, rxBytesCount is the reply. + stat := benchmark_runner.NewStat().AddEntry([]byte(cmdType), []byte(cmdQueryId), uint64(t.Unix()), took, hadError, isTimeout, rxBytesCount, txBytesCount) p.cmdChan <- *stat } cmds = nil @@ -386,8 +388,13 @@ func preProcessCmd(row string) (cmdType string, cmdQueryId string, keyPos int, c if initialPos >= 0 { clusterSlot = int(radix.ClusterSlot([]byte(key))) } - // Subtract the base64 shrink so byte accounting reflects the decoded - // bytes actually sent to Redis, not the larger base64 text in the row. + // bytelen approximates the sent (TX) payload: the row minus the leading + // cmdType label, minus the base64 shrink (so it reflects decoded bytes, + // not the larger base64 text). It is an application-payload proxy, not + // exact RESP wire bytes — it still counts CSV separators / the queryId + // and pos columns and omits RESP framing (*N\r\n, per-arg $len\r\n). The + // error is negligible for large payloads (e.g. vector blobs) but can be + // sizable for many-tiny-arg commands. bytelen = uint64(len(row)) - uint64(len(cmdType)) - shrink } else { err = fmt.Errorf("input string does not have the minimum required size of 2: %s", row) diff --git a/cmd/ftsb_redisearch/send_stats_test.go b/cmd/ftsb_redisearch/send_stats_test.go new file mode 100644 index 0000000..cbd3e09 --- /dev/null +++ b/cmd/ftsb_redisearch/send_stats_test.go @@ -0,0 +1,136 @@ +package main + +import ( + "errors" + "testing" + "time" + + "github.com/RediSearch/ftsb/benchmark_runner" + radix "github.com/mediocregopher/radix/v3" +) + +// fakeClient is a radix.Client whose Do returns a canned error (nil = success), +// so the recording path in sendFlatCmd/sendIfRequired can be exercised without +// a real Redis. +type fakeClient struct { + calls int + err error +} + +func (f *fakeClient) Do(a radix.Action) error { f.calls++; return f.err } +func (f *fakeClient) Close() error { return nil } + +// Regression guard for issue #111: the bytes we SEND to Redis (txBytesCount) +// must be recorded as Tx(), not Rx(). Before the fix the AddEntry arguments +// were swapped, so sent bytes were reported under RxBytes and TxBytes was 0. +func TestSendFlatCmdRecordsSentBytesAsTx(t *testing.T) { + // pipeline=1 forces sendIfRequired to flush on the first command. Set it + // explicitly (rather than trusting the flag default) so a stray global + // left by another test can't make the <-p.cmdChan receive block forever. + savedPipeline := pipeline + pipeline = 1 + defer func() { pipeline = savedPipeline }() + + p := &processor{cmdChan: make(chan benchmark_runner.Stat, 1)} + const txBytesCount = uint64(4096) // request/sent bytes for this command + + _, _, hadError := sendFlatCmd( + p, &fakeClient{}, "WRITE", "w1", "HSET", + []string{"doc:1", "vec", "payload"}, txBytesCount, + make([]radix.CmdAction, 0), make([]interface{}, 0), make([]time.Time, 0), + ) + if hadError { + t.Fatal("unexpected error from fake client") + } + + stat := <-p.cmdChan + entries := stat.CmdStats() + if len(entries) != 1 { + t.Fatalf("expected 1 stat entry, got %d", len(entries)) + } + if got := entries[0].Tx(); got != txBytesCount { + t.Fatalf("Tx() = %d, want %d (sent bytes must land in Tx, not Rx)", got, txBytesCount) + } + // Replies are discarded (rcv is a nil interface), so received bytes are 0. + if got := entries[0].Rx(); got != 0 { + t.Fatalf("Rx() = %d, want 0 (replies are not captured)", got) + } +} + +func TestGetRxLen(t *testing.T) { + if got := getRxLen("abc"); got != 3 { + t.Fatalf("getRxLen(string) = %d, want 3", got) + } + if got := getRxLen([]string{"ab", "cde"}); got != 5 { + t.Fatalf("getRxLen([]string) = %d, want 5", got) + } + if got := getRxLen(nil); got != 0 { + t.Fatalf("getRxLen(nil) = %d, want 0", got) + } + if got := getRxLen(42); got != 0 { + t.Fatalf("getRxLen(int) = %d, want 0", got) + } +} + +// On a command error (continue-on-error), a stat must still be recorded, marked +// as an error, and carry the correct sent-byte count in Tx(). +func TestSendFlatCmdRecordsErrorStatWithCorrectTx(t *testing.T) { + savedPipeline, savedContinue := pipeline, continueOnErr + pipeline, continueOnErr = 1, true + defer func() { pipeline, continueOnErr = savedPipeline, savedContinue }() + + p := &processor{cmdChan: make(chan benchmark_runner.Stat, 1)} + const txBytesCount = uint64(128) + + _, _, hadError := sendFlatCmd( + p, &fakeClient{err: errors.New("connection refused")}, "WRITE", "w1", "HSET", + []string{"doc:1"}, txBytesCount, + make([]radix.CmdAction, 0), make([]interface{}, 0), make([]time.Time, 0), + ) + if !hadError { + t.Fatal("expected hadError=true when client.Do returns an error") + } + + stat := <-p.cmdChan + entries := stat.CmdStats() + if len(entries) != 1 { + t.Fatalf("expected 1 stat entry, got %d", len(entries)) + } + if !entries[0].Error() { + t.Fatal("entry should be marked as an error") + } + if got := entries[0].Tx(); got != txBytesCount { + t.Fatalf("Tx() = %d, want %d (sent bytes recorded even on error)", got, txBytesCount) + } +} + +// An i/o timeout error must set the timedOut flag on the recorded stat. +func TestSendFlatCmdMarksTimeout(t *testing.T) { + savedPipeline, savedContinue := pipeline, continueOnErr + pipeline, continueOnErr = 1, true + defer func() { pipeline, continueOnErr = savedPipeline, savedContinue }() + + p := &processor{cmdChan: make(chan benchmark_runner.Stat, 1)} + _, _, hadError := sendFlatCmd( + p, &fakeClient{err: errors.New("dial tcp 127.0.0.1:6379: i/o timeout")}, "READ", "r1", "FT.SEARCH", + []string{"idx"}, 64, + make([]radix.CmdAction, 0), make([]interface{}, 0), make([]time.Time, 0), + ) + if !hadError { + t.Fatal("expected hadError=true on timeout") + } + stat := <-p.cmdChan + entries := stat.CmdStats() + if len(entries) != 1 { + t.Fatalf("expected 1 stat entry, got %d", len(entries)) + } + if !entries[0].TimedOut() { + t.Fatal("entry should be marked as timed out for an i/o timeout error") + } + if !entries[0].Error() { + t.Fatal("a timeout is also an error") + } + if got := entries[0].Tx(); got != 64 { + t.Fatalf("Tx() = %d, want 64 (sent bytes recorded even on timeout)", got) + } +}