Merge branch 'main' into feature/reduce-cpu

Author: bootjp

.gitignore

@@ -37,3 +37,4 @@ jepsen/.ssh/
 # Build and lint cache directories
 .cache/
 .golangci-cache/
+server

adapter/redis_lua_context.go

@@ -486,7 +486,12 @@ func (c *luaScriptContext) materializeList(key []byte, st *luaListState) error {
 		return nil
 	}
 
-	values := make([]string, 0, st.currentLen())
+	length := st.currentLen()
+	maxInt64 := int64(math.MaxInt)
+	if length > maxInt64 {
+		length = maxInt64
+	}
+	values := make([]string, 0, int(length))
 	values = append(values, st.leftValues...)
 	if remaining := st.remainingOriginalLen(); remaining > 0 {
 		start := st.leftTrim

cmd/server/demo.go

@@ -38,6 +38,8 @@ var (
 	dynamoAddress  = flag.String("dynamoAddress", ":8000", "DynamoDB-compatible API address")
 	metricsAddress = flag.String("metricsAddress", "127.0.0.1:9090", "Prometheus metrics address")
 	metricsToken   = flag.String("metricsToken", "", "Bearer token for Prometheus metrics; required for non-loopback metricsAddress")
+	pprofAddress   = flag.String("pprofAddress", "localhost:6060", "TCP host+port for pprof debug endpoints; empty to disable")
+	pprofToken     = flag.String("pprofToken", "", "Bearer token for pprof; required for non-loopback pprofAddress")
 	raftID         = flag.String("raftId", "", "Raft ID")
 	raftDataDir    = flag.String("raftDataDir", "/var/lib/elastickv", "Raft data directory")
 	raftBootstrap  = flag.Bool("raftBootstrap", false, "Bootstrap cluster")
@@ -67,6 +69,8 @@ type config struct {
 	dynamoAddress  string
 	metricsAddress string
 	metricsToken   string
+	pprofAddress   string
+	pprofToken     string
 	raftID         string
 	raftDataDir    string
 	raftBootstrap  bool
@@ -86,6 +90,8 @@ func main() {
 			dynamoAddress:  *dynamoAddress,
 			metricsAddress: *metricsAddress,
 			metricsToken:   *metricsToken,
+			pprofAddress:   *pprofAddress,
+			pprofToken:     *pprofToken,
 			raftID:         *raftID,
 			raftDataDir:    *raftDataDir,
 			raftBootstrap:  *raftBootstrap,
@@ -106,6 +112,7 @@ func main() {
 				dynamoAddress:  "127.0.0.1:63801",
 				metricsAddress: "0.0.0.0:9091",
 				metricsToken:   demoMetricsToken,
+				pprofAddress:   "127.0.0.1:6061",
 				raftID:         "n1",
 				raftDataDir:    "", // In-memory
 				raftBootstrap:  true,
@@ -116,6 +123,7 @@ func main() {
 				dynamoAddress:  "127.0.0.1:63802",
 				metricsAddress: "0.0.0.0:9092",
 				metricsToken:   demoMetricsToken,
+				pprofAddress:   "127.0.0.1:6062",
 				raftID:         "n2",
 				raftDataDir:    "",
 				raftBootstrap:  false,
@@ -126,6 +134,7 @@ func main() {
 				dynamoAddress:  "127.0.0.1:63803",
 				metricsAddress: "0.0.0.0:9093",
 				metricsToken:   demoMetricsToken,
+				pprofAddress:   "127.0.0.1:6063",
 				raftID:         "n3",
 				raftDataDir:    "",
 				raftBootstrap:  false,
@@ -443,15 +452,10 @@ func run(ctx context.Context, eg *errgroup.Group, cfg config) error {
 		adapter.WithDynamoDBRequestObserver(metricsRegistry.DynamoDBObserver()),
 	)
 	cleanup.Add(ds.Stop)
-	metricsL, ms, err := setupMetricsHTTPServer(ctx, lc, cfg.metricsAddress, cfg.metricsToken, metricsRegistry.Handler())
+	metricsL, ms, pprofL, ps, err := setupObservabilityServers(ctx, lc, &cleanup, cfg, metricsRegistry.Handler())
 	if err != nil {
 		return err
 	}
-	if metricsL != nil {
-		cleanup.Add(func() {
-			_ = metricsL.Close()
-		})
-	}
 
 	eg.Go(catalogWatcherTask(ctx, distCatalog, distEngine))
 	eg.Go(func() error { return compactor.Run(ctx) })
@@ -463,11 +467,31 @@ func run(ctx context.Context, eg *errgroup.Group, cfg config) error {
 	eg.Go(dynamoServeTask(ds, cfg.dynamoAddress))
 	eg.Go(monitoring.MetricsShutdownTask(ctx, ms, cfg.metricsAddress))
 	eg.Go(monitoring.MetricsServeTask(ms, metricsL, cfg.metricsAddress))
+	eg.Go(monitoring.PprofShutdownTask(ctx, ps, cfg.pprofAddress))
+	eg.Go(monitoring.PprofServeTask(ps, pprofL, cfg.pprofAddress))
 
 	cleanup.Release()
 	return nil
 }
 
+func setupObservabilityServers(ctx context.Context, lc net.ListenConfig, cleanup *internalutil.CleanupStack, cfg config, metricsHandler http.Handler) (metricsL net.Listener, ms *http.Server, pprofL net.Listener, ps *http.Server, err error) {
+	metricsL, ms, err = setupMetricsHTTPServer(ctx, lc, cfg.metricsAddress, cfg.metricsToken, metricsHandler)
+	if err != nil {
+		return
+	}
+	if metricsL != nil {
+		cleanup.Add(func() { _ = metricsL.Close() })
+	}
+	pprofL, ps, err = setupPprofHTTPServer(ctx, lc, cfg.pprofAddress, cfg.pprofToken)
+	if err != nil {
+		return
+	}
+	if pprofL != nil {
+		cleanup.Add(func() { _ = pprofL.Close() })
+	}
+	return
+}
+
 func setupMetricsHTTPServer(ctx context.Context, lc net.ListenConfig, metricsAddress string, metricsToken string, handler http.Handler) (net.Listener, *http.Server, error) {
 	metricsAddress = strings.TrimSpace(metricsAddress)
 	if metricsAddress == "" || handler == nil {
@@ -476,7 +500,7 @@ func setupMetricsHTTPServer(ctx context.Context, lc net.ListenConfig, metricsAdd
 	if _, _, err := net.SplitHostPort(metricsAddress); err != nil {
 		return nil, nil, errors.Wrapf(err, "invalid metricsAddress %q", metricsAddress)
 	}
-	if monitoring.MetricsAddressRequiresToken(metricsAddress) && strings.TrimSpace(metricsToken) == "" {
+	if monitoring.AddressRequiresToken(metricsAddress) && strings.TrimSpace(metricsToken) == "" {
 		return nil, nil, errors.New("metricsToken is required when metricsAddress is not loopback")
 	}
 	metricsL, err := lc.Listen(ctx, "tcp", metricsAddress)
@@ -487,6 +511,25 @@ func setupMetricsHTTPServer(ctx context.Context, lc net.ListenConfig, metricsAdd
 	return metricsL, ms, nil
 }
 
+func setupPprofHTTPServer(ctx context.Context, lc net.ListenConfig, pprofAddress string, pprofToken string) (net.Listener, *http.Server, error) {
+	pprofAddress = strings.TrimSpace(pprofAddress)
+	if pprofAddress == "" {
+		return nil, nil, nil
+	}
+	if _, _, err := net.SplitHostPort(pprofAddress); err != nil {
+		return nil, nil, errors.Wrapf(err, "invalid pprofAddress %q", pprofAddress)
+	}
+	if monitoring.AddressRequiresToken(pprofAddress) && strings.TrimSpace(pprofToken) == "" {
+		return nil, nil, errors.New("pprofToken is required when pprofAddress is not loopback")
+	}
+	pprofL, err := lc.Listen(ctx, "tcp", pprofAddress)
+	if err != nil {
+		return nil, nil, errors.WithStack(err)
+	}
+	ps := monitoring.NewPprofServer(pprofToken)
+	return pprofL, ps, nil
+}
+
 func bootstrapClusterIfNeeded(r *raft.Raft, cfg config) error {
 	if !cfg.raftBootstrap {
 		return nil

cmd/server/demo_test.go

@@ -27,3 +27,17 @@ func TestSetupMetricsHTTPServerRejectsInvalidAddressBeforeTokenCheck(t *testing.
 	require.Nil(t, listener)
 	require.Nil(t, server)
 }
+
+func TestSetupPprofHTTPServerAllowsBlankAddress(t *testing.T) {
+	listener, server, err := setupPprofHTTPServer(context.Background(), net.ListenConfig{}, "", "")
+	require.NoError(t, err)
+	require.Nil(t, listener)
+	require.Nil(t, server)
+}
+
+func TestSetupPprofHTTPServerRejectsInvalidAddressBeforeTokenCheck(t *testing.T) {
+	listener, server, err := setupPprofHTTPServer(context.Background(), net.ListenConfig{}, "localhost", "")
+	require.ErrorContains(t, err, `invalid pprofAddress "localhost"`)
+	require.Nil(t, listener)
+	require.Nil(t, server)
+}

kv/txn_errors.go

@@ -39,9 +39,9 @@ func TxnLockedDetails(err error) ([]byte, string, bool) {
 
 func (e *TxnLockedError) Error() string {
 	if e.detail != "" {
-		return fmt.Sprintf("key: %s (%s): %s", string(e.key), e.detail, ErrTxnLocked)
+		return fmt.Sprintf("key: %s (%s): %v", string(e.key), e.detail, ErrTxnLocked)
 	}
-	return fmt.Sprintf("key: %s: %s", string(e.key), ErrTxnLocked)
+	return fmt.Sprintf("key: %s: %v", string(e.key), ErrTxnLocked)
 }
 
 func (e *TxnLockedError) Unwrap() error {

main.go

@@ -39,6 +39,8 @@ var (
 	dynamoAddr           = flag.String("dynamoAddress", "localhost:8000", "TCP host+port for DynamoDB-compatible API")
 	metricsAddr          = flag.String("metricsAddress", "localhost:9090", "TCP host+port for Prometheus metrics")
 	metricsToken         = flag.String("metricsToken", "", "Bearer token for Prometheus metrics; required for non-loopback metricsAddress")
+	pprofAddr            = flag.String("pprofAddress", "localhost:6060", "TCP host+port for pprof debug endpoints; empty to disable")
+	pprofToken           = flag.String("pprofToken", "", "Bearer token for pprof; required for non-loopback pprofAddress")
 	raftId               = flag.String("raftId", "", "Node id used by Raft")
 	raftDir              = flag.String("raftDataDir", "data/", "Raft data dir")
 	raftBootstrap        = flag.Bool("raftBootstrap", false, "Whether to bootstrap the Raft cluster")
@@ -133,6 +135,8 @@ func run() error {
 		dynamoAddress:   *dynamoAddr,
 		metricsAddress:  *metricsAddr,
 		metricsToken:    *metricsToken,
+		pprofAddress:    *pprofAddr,
+		pprofToken:      *pprofToken,
 		metricsRegistry: metricsRegistry,
 	}
 	if err := runner.start(); err != nil {
@@ -415,6 +419,27 @@ func startDynamoDBServer(ctx context.Context, lc *net.ListenConfig, eg *errgroup
 	return nil
 }
 
+func startPprofServer(ctx context.Context, lc *net.ListenConfig, eg *errgroup.Group, pprofAddr string, pprofToken string) error {
+	pprofAddr = strings.TrimSpace(pprofAddr)
+	if pprofAddr == "" {
+		return nil
+	}
+	if _, _, err := net.SplitHostPort(pprofAddr); err != nil {
+		return errors.Wrapf(err, "invalid pprofAddress %q; expected host:port", pprofAddr)
+	}
+	if monitoring.AddressRequiresToken(pprofAddr) && strings.TrimSpace(pprofToken) == "" {
+		return errors.New("pprofToken is required when pprofAddress is not loopback")
+	}
+	pprofL, err := lc.Listen(ctx, "tcp", pprofAddr)
+	if err != nil {
+		return errors.Wrapf(err, "failed to listen on %s", pprofAddr)
+	}
+	pprofServer := monitoring.NewPprofServer(pprofToken)
+	eg.Go(monitoring.PprofShutdownTask(ctx, pprofServer, pprofAddr))
+	eg.Go(monitoring.PprofServeTask(pprofServer, pprofL, pprofAddr))
+	return nil
+}
+
 func startMetricsServer(ctx context.Context, lc *net.ListenConfig, eg *errgroup.Group, metricsAddr string, metricsToken string, handler http.Handler) error {
 	metricsAddr = strings.TrimSpace(metricsAddr)
 	if metricsAddr == "" || handler == nil {
@@ -423,7 +448,7 @@ func startMetricsServer(ctx context.Context, lc *net.ListenConfig, eg *errgroup.
 	if _, _, err := net.SplitHostPort(metricsAddr); err != nil {
 		return errors.Wrapf(err, "invalid metricsAddress %q; expected host:port", metricsAddr)
 	}
-	if monitoring.MetricsAddressRequiresToken(metricsAddr) && strings.TrimSpace(metricsToken) == "" {
+	if monitoring.AddressRequiresToken(metricsAddr) && strings.TrimSpace(metricsToken) == "" {
 		return errors.New("metricsToken is required when metricsAddress is not loopback")
 	}
 	metricsL, err := lc.Listen(ctx, "tcp", metricsAddr)
@@ -516,6 +541,8 @@ type runtimeServerRunner struct {
 	dynamoAddress   string
 	metricsAddress  string
 	metricsToken    string
+	pprofAddress    string
+	pprofToken      string
 	metricsRegistry *monitoring.Registry
 }
 
@@ -532,5 +559,8 @@ func (r runtimeServerRunner) start() error {
 	if err := startMetricsServer(r.ctx, r.lc, r.eg, r.metricsAddress, r.metricsToken, r.metricsRegistry.Handler()); err != nil {
 		return waitErrgroupAfterStartupFailure(r.cancel, r.eg, err)
 	}
+	if err := startPprofServer(r.ctx, r.lc, r.eg, r.pprofAddress, r.pprofToken); err != nil {
+		return waitErrgroupAfterStartupFailure(r.cancel, r.eg, err)
+	}
 	return nil
 }

main_pprof_test.go

@@ -0,0 +1,22 @@
+package main
+
+import (
+	"context"
+	"net"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+	"golang.org/x/sync/errgroup"
+)
+
+func TestStartPprofServerRejectsInvalidAddressBeforeTokenCheck(t *testing.T) {
+	eg, ctx := errgroup.WithContext(context.Background())
+	err := startPprofServer(ctx, &net.ListenConfig{}, eg, "localhost", "")
+	require.ErrorContains(t, err, `invalid pprofAddress "localhost"`)
+}
+
+func TestStartPprofServerAllowsEmptyAddress(t *testing.T) {
+	eg, ctx := errgroup.WithContext(context.Background())
+	err := startPprofServer(ctx, &net.ListenConfig{}, eg, "", "")
+	require.NoError(t, err)
+}