mirror of
https://github.com/hanzokv/go.git
synced 2026-09-24 00:31:51 +00:00
zeekay
6994cf05aa
We do not use Redis. The client was a go-redis fork and still said so
everywhere: `package redis`, `redis.NewClient`, `redis://`, `extra/redisotel`.
Consumers wrote `redis.Client` against a Valkey-backed store called KV.
package redis -> package kv (93 decls)
redis.X -> kv.X (2,794 call sites)
"redis: …" errors -> "kv: …" (206)
extra/redis{otel,cmd,prometheus,census} -> extra/kv{otel,cmd,prometheus,census}
redis:// , rediss:// -> kv:// , kvs:// (ParseURL)
Redis*/*Redis* idents -> KV*/*KV*
REDIS_VERSION, REDIS_ENDPOINTS_CONFIG_PATH -> KV_*
I previously argued this rename was unsafe because much of it was "wire
contract". That was wrong, and the categorisation shows why: RESP command names
are GET/SET/HSET — they never contain the word. Of 5,771 occurrences only two
kinds were server-visible at all: the ParseURL scheme (we choose what we pass;
our configs already use kv://) and the CLIENT SETINFO lib-name label, which the
server merely records. Nothing here changes a byte of protocol.
Two things deliberately keep the old spelling:
RediSearch — names a specific external module the FT.* commands map to.
StoreDist — contains the letters by accident (sto-reDis-t), not the word.
Breaking for consumers by design: every import site must move redis.X -> kv.X.
Minor bump, never a major — this module stays v9 forever.
143 lines
3.1 KiB
Go
143 lines
3.1 KiB
Go
package e2e
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"strings"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"github.com/hanzokv/go/v9"
|
|
)
|
|
|
|
type CommandRunnerStats struct {
|
|
Operations int64
|
|
Errors int64
|
|
TimeoutErrors int64
|
|
ErrorsList []error
|
|
}
|
|
|
|
// CommandRunner provides utilities for running commands during tests
|
|
type CommandRunner struct {
|
|
executing atomic.Bool
|
|
client kv.UniversalClient
|
|
stopCh chan struct{}
|
|
operationCount atomic.Int64
|
|
errorCount atomic.Int64
|
|
timeoutErrors atomic.Int64
|
|
errors []error
|
|
errorsMutex sync.Mutex
|
|
}
|
|
|
|
// NewCommandRunner creates a new command runner
|
|
func NewCommandRunner(client kv.UniversalClient) (*CommandRunner, func()) {
|
|
stopCh := make(chan struct{})
|
|
return &CommandRunner{
|
|
client: client,
|
|
stopCh: stopCh,
|
|
errors: make([]error, 0),
|
|
}, func() {
|
|
stopCh <- struct{}{}
|
|
}
|
|
}
|
|
|
|
func (cr *CommandRunner) Stop() {
|
|
select {
|
|
case cr.stopCh <- struct{}{}:
|
|
return
|
|
case <-time.After(500 * time.Millisecond):
|
|
return
|
|
}
|
|
}
|
|
|
|
func (cr *CommandRunner) Close() {
|
|
close(cr.stopCh)
|
|
}
|
|
|
|
// FireCommandsUntilStop runs commands continuously until stop signal
|
|
func (cr *CommandRunner) FireCommandsUntilStop(ctx context.Context) {
|
|
if !cr.executing.CompareAndSwap(false, true) {
|
|
return
|
|
}
|
|
defer cr.executing.Store(false)
|
|
fmt.Printf("[CR] Starting command runner...\n")
|
|
defer fmt.Printf("[CR] Command runner stopped\n")
|
|
// High frequency for timeout testing
|
|
ticker := time.NewTicker(100 * time.Millisecond)
|
|
defer ticker.Stop()
|
|
|
|
counter := 0
|
|
for {
|
|
select {
|
|
case <-cr.stopCh:
|
|
return
|
|
case <-ctx.Done():
|
|
return
|
|
case <-ticker.C:
|
|
poolSize := cr.client.PoolStats().IdleConns
|
|
if poolSize == 0 {
|
|
poolSize = 1
|
|
}
|
|
wg := sync.WaitGroup{}
|
|
for i := 0; i < int(poolSize); i++ {
|
|
wg.Add(1)
|
|
go func(i int) {
|
|
defer wg.Done()
|
|
key := fmt.Sprintf("timeout-test-key-%d-%d", counter, i)
|
|
value := fmt.Sprintf("timeout-test-value-%d-%d", counter, i)
|
|
|
|
// Use a short timeout context for individual operations
|
|
opCtx, cancel := context.WithTimeout(ctx, 2*time.Second)
|
|
err := cr.client.Set(opCtx, key, value, time.Minute).Err()
|
|
cancel()
|
|
|
|
cr.operationCount.Add(1)
|
|
if err != nil {
|
|
if err == kv.ErrClosed || strings.Contains(err.Error(), "client is closed") {
|
|
select {
|
|
case <-cr.stopCh:
|
|
return
|
|
default:
|
|
}
|
|
return
|
|
}
|
|
|
|
fmt.Printf("Error: %v\n", err)
|
|
cr.errorCount.Add(1)
|
|
|
|
// Check if it's a timeout error
|
|
if isTimeoutError(err) {
|
|
cr.timeoutErrors.Add(1)
|
|
}
|
|
|
|
cr.errorsMutex.Lock()
|
|
cr.errors = append(cr.errors, err)
|
|
cr.errorsMutex.Unlock()
|
|
}
|
|
}(i)
|
|
}
|
|
wg.Wait()
|
|
counter++
|
|
}
|
|
}
|
|
}
|
|
|
|
// GetStats returns operation statistics
|
|
func (cr *CommandRunner) GetStats() CommandRunnerStats {
|
|
cr.errorsMutex.Lock()
|
|
defer cr.errorsMutex.Unlock()
|
|
|
|
errorList := make([]error, len(cr.errors))
|
|
copy(errorList, cr.errors)
|
|
|
|
stats := CommandRunnerStats{
|
|
Operations: cr.operationCount.Load(),
|
|
Errors: cr.errorCount.Load(),
|
|
TimeoutErrors: cr.timeoutErrors.Load(),
|
|
ErrorsList: errorList,
|
|
}
|
|
|
|
return stats
|
|
}
|