Merge pull request #8817 from BramGruneir/8473

server: enable test cluster to take different store configurations

acceptance/cluster/localcluster.go

@@ -44,7 +44,6 @@ import (
 	roachClient "github.com/cockroachdb/cockroach/internal/client"
 	"github.com/cockroachdb/cockroach/rpc"
 	"github.com/cockroachdb/cockroach/security"
-	"github.com/cockroachdb/cockroach/server"
 	"github.com/cockroachdb/cockroach/util"
 	"github.com/cockroachdb/cockroach/util/log"
 	"github.com/cockroachdb/cockroach/util/stop"
@@ -462,7 +461,7 @@ func (l *LocalCluster) startNode(node *testNode) {
 	}
 
 	for _, store := range node.stores {
-		storeSpec := server.StoreSpec{
+		storeSpec := base.StoreSpec{
 			Path:        store.dataStr,
 			SizeInBytes: int64(store.config.MaxRanges) * maxRangeBytes,
 		}

server/store_spec.go → base/store_spec.go

@@ -14,7 +14,7 @@
 //
 // Author: Bram Gruneir ([email protected])
 
-package server
+package base
 
 import (
 	"bytes"
@@ -26,15 +26,18 @@ import (
 
 	"github.com/dustin/go-humanize"
 
-	"github.com/cockroachdb/cockroach/config"
 	"github.com/cockroachdb/cockroach/roachpb"
 	"github.com/cockroachdb/cockroach/util/humanizeutil"
 )
 
 // This file implements method receivers for members of server.Context struct
 // -- 'Stores' and 'JoinList', which satisfies pflag's value interface
 
-var minimumStoreSize = 10 * config.DefaultZoneConfig().RangeMaxBytes
+// MinimumStoreSize is the smallest size in bytes that a store can have. This
+// number is based on config's defaultZoneConfig's RangeMaxBytes, which is
+// extremely stable. To avoid adding the dependency on config here, it is just
+// hard coded to 640MiB.
+const MinimumStoreSize = 10 * 64 << 20
 
 // StoreSpec contains the details that can be specified in the cli pertaining
 // to the --store flag.
@@ -166,9 +169,9 @@ func newStoreSpec(value string) (StoreSpec, error) {
 				if err != nil {
 					return StoreSpec{}, fmt.Errorf("could not parse store size (%s) %s", value, err)
 				}
-				if ss.SizeInBytes < minimumStoreSize {
+				if ss.SizeInBytes < MinimumStoreSize {
 					return StoreSpec{}, fmt.Errorf("store size (%s) must be larger than %s", value,
-						humanizeutil.IBytes(minimumStoreSize))
+						humanizeutil.IBytes(MinimumStoreSize))
 				}
 			}
 		case "attrs":

server/store_spec_test.go → base/store_spec_test.go

@@ -14,7 +14,7 @@
 //
 // Author: Bram Gruneir ([email protected])
 
-package server
+package base
 
 import (
 	"fmt"

base/test_server_args.go

@@ -41,7 +41,11 @@ type TestServerArgs struct {
 	// JoinAddr (if nonempty) is the address of a node we are joining.
 	JoinAddr string
 
-	StoresPerNode int
+	// StoreSpecs define the stores for this server. If you want more than one
+	// store per node, populate this array with StoreSpecs each representing a
+	// store. If no StoreSpecs are provided than a single DefaultTestStoreSpec
+	// will be used.
+	StoreSpecs []StoreSpec
 
 	// Fields copied to the server.Context.
 	Insecure              bool
@@ -74,6 +78,19 @@ type TestClusterArgs struct {
 	ServerArgs TestServerArgs
 	// ReplicationMode controls how replication is to be done in the cluster.
 	ReplicationMode TestClusterReplicationMode
+
+	// ServerArgsPerNode override the default ServerArgs with the value in this
+	// map. The map's key is an index within TestCluster.Servers. If there is
+	// no entry in the map for a particular server, the default ServerArgs are
+	// used.
+	ServerArgsPerNode map[int]TestServerArgs
+}
+
+// DefaultTestStoreSpec is just a single in memory store of 100 MiB with no
+// special attributes.
+var DefaultTestStoreSpec = StoreSpec{
+	SizeInBytes: 100 << 20,
+	InMemory:    true,
 }
 
 // TestClusterReplicationMode represents the replication settings for a TestCluster.

kv/dist_sender_server_test.go

@@ -1162,7 +1162,12 @@ func TestPropagateTxnOnPushError(t *testing.T) {
 // should be willing to get rid of it.
 func TestRequestToUninitializedRange(t *testing.T) {
 	defer leaktest.AfterTest(t)()
-	srv, _, _ := serverutils.StartServer(t, base.TestServerArgs{StoresPerNode: 2})
+	srv, _, _ := serverutils.StartServer(t, base.TestServerArgs{
+		StoreSpecs: []base.StoreSpec{
+			base.DefaultTestStoreSpec,
+			base.DefaultTestStoreSpec,
+		},
+	})
 	defer srv.Stopper().Stop()
 	s := srv.(*server.TestServer)
 

server/context.go

@@ -76,7 +76,7 @@ type Context struct {
 	SocketFile string
 
 	// Stores is specified to enable durable key-value storage.
-	Stores StoreSpecList
+	Stores base.StoreSpecList
 
 	// Attrs specifies a colon-separated list of node topography or machine
 	// capabilities, used to match capabilities or location preferences specified
@@ -86,7 +86,7 @@ type Context struct {
 	// JoinList is a list of node addresses that act as bootstrap hosts for
 	// connecting to the gossip network. Each item in the list can actually be
 	// multiple comma-separated addresses, kept for backward-compatibility.
-	JoinList JoinListType
+	JoinList base.JoinListType
 
 	// CacheSize is the amount of memory in bytes to use for caching data.
 	// The value is split evenly between the stores if there are more than one.
@@ -337,8 +337,8 @@ func MakeContext() Context {
 		MetricsSampleInterval:    defaultMetricsSampleInterval,
 		TimeUntilStoreDead:       defaultTimeUntilStoreDead,
 		ReservationsEnabled:      defaultReservationsEnabled,
-		Stores: StoreSpecList{
-			Specs: []StoreSpec{{Path: defaultStorePath}},
+		Stores: base.StoreSpecList{
+			Specs: []base.StoreSpec{{Path: defaultStorePath}},
 		},
 	}
 	ctx.Context.InitDefaults()
@@ -371,9 +371,9 @@ func (ctx *Context) InitStores(stopper *stop.Stopper) error {
 				}
 				sizeInBytes = int64(float64(sysMem) * spec.SizePercent / 100)
 			}
-			if sizeInBytes != 0 && sizeInBytes < minimumStoreSize {
+			if sizeInBytes != 0 && sizeInBytes < base.MinimumStoreSize {
 				return fmt.Errorf("%f%% of memory is only %s bytes, which is below the minimum requirement of %s",
-					spec.SizePercent, humanizeutil.IBytes(sizeInBytes), humanizeutil.IBytes(minimumStoreSize))
+					spec.SizePercent, humanizeutil.IBytes(sizeInBytes), humanizeutil.IBytes(base.MinimumStoreSize))
 			}
 			ctx.Engines = append(ctx.Engines, engine.NewInMem(spec.Attributes, sizeInBytes, stopper))
 		} else {
@@ -384,9 +384,9 @@ func (ctx *Context) InitStores(stopper *stop.Stopper) error {
 				}
 				sizeInBytes = int64(float64(fileSystemUsage.Total) * spec.SizePercent / 100)
 			}
-			if sizeInBytes != 0 && sizeInBytes < minimumStoreSize {
+			if sizeInBytes != 0 && sizeInBytes < base.MinimumStoreSize {
 				return fmt.Errorf("%f%% of %s's total free space is only %s bytes, which is below the minimum requirement of %s",
-					spec.SizePercent, spec.Path, humanizeutil.IBytes(sizeInBytes), humanizeutil.IBytes(minimumStoreSize))
+					spec.SizePercent, spec.Path, humanizeutil.IBytes(sizeInBytes), humanizeutil.IBytes(base.MinimumStoreSize))
 			}
 			ctx.Engines = append(
 				ctx.Engines,

server/context_test.go

@@ -22,6 +22,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/cockroachdb/cockroach/base"
 	"github.com/cockroachdb/cockroach/gossip/resolver"
 	"github.com/cockroachdb/cockroach/util/envutil"
 	"github.com/cockroachdb/cockroach/util/leaktest"
@@ -32,7 +33,7 @@ func TestParseInitNodeAttributes(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	ctx := MakeContext()
 	ctx.Attrs = "attr1=val1::attr2=val2"
-	ctx.Stores = StoreSpecList{Specs: []StoreSpec{{InMemory: true, SizeInBytes: minimumStoreSize * 100}}}
+	ctx.Stores = base.StoreSpecList{Specs: []base.StoreSpec{{InMemory: true, SizeInBytes: base.MinimumStoreSize * 100}}}
 	stopper := stop.NewStopper()
 	defer stopper.Stop()
 	if err := ctx.InitStores(stopper); err != nil {
@@ -53,7 +54,7 @@ func TestParseJoinUsingAddrs(t *testing.T) {
 	defer leaktest.AfterTest(t)()
 	ctx := MakeContext()
 	ctx.JoinList = []string{"localhost:12345,,localhost:23456", "localhost:34567"}
-	ctx.Stores = StoreSpecList{Specs: []StoreSpec{{InMemory: true, SizeInBytes: minimumStoreSize * 100}}}
+	ctx.Stores = base.StoreSpecList{Specs: []base.StoreSpec{{InMemory: true, SizeInBytes: base.MinimumStoreSize * 100}}}
 	stopper := stop.NewStopper()
 	defer stopper.Stop()
 	if err := ctx.InitStores(stopper); err != nil {

server/status_test.go

@@ -151,7 +151,13 @@ func TestStatusGossipJson(t *testing.T) {
 // the scan to complete, and return the server. The caller is
 // responsible for stopping the server.
 func startServer(t *testing.T) *TestServer {
-	tsI, _, kvDB := serverutils.StartServer(t, base.TestServerArgs{StoresPerNode: 3})
+	tsI, _, kvDB := serverutils.StartServer(t, base.TestServerArgs{
+		StoreSpecs: []base.StoreSpec{
+			base.DefaultTestStoreSpec,
+			base.DefaultTestStoreSpec,
+			base.DefaultTestStoreSpec,
+		},
+	})
 
 	ts := tsI.(*TestServer)
 

server/testserver.go

@@ -23,6 +23,7 @@ import (
 	"path/filepath"
 	"time"
 
+	"github.com/pkg/errors"
 	"golang.org/x/net/context"
 
 	"github.com/cockroachdb/cockroach/base"
@@ -41,7 +42,6 @@ import (
 	"github.com/cockroachdb/cockroach/util/hlc"
 	"github.com/cockroachdb/cockroach/util/metric"
 	"github.com/cockroachdb/cockroach/util/stop"
-	"github.com/pkg/errors"
 )
 
 const (
@@ -218,11 +218,24 @@ func (ts *TestServer) Start(params base.TestServerArgs) error {
 
 	// Ensure we have the correct number of engines. Add in-memory ones where
 	// needed. There must be at least one store/engine.
-	if params.StoresPerNode < 1 {
-		params.StoresPerNode = 1
+	if len(params.StoreSpecs) == 0 {
+		params.StoreSpecs = []base.StoreSpec{base.DefaultTestStoreSpec}
 	}
-	for i := len(ts.Ctx.Engines); i < params.StoresPerNode; i++ {
-		ts.Ctx.Engines = append(ts.Ctx.Engines, engine.NewInMem(roachpb.Attributes{}, 100<<20, params.Stopper))
+	for _, storeSpec := range params.StoreSpecs {
+		if storeSpec.InMemory {
+			if storeSpec.SizePercent > 0 {
+				panic(fmt.Sprintf("test server does not yet support in memory stores based on percentage of total memory: %s", storeSpec))
+			}
+			ts.Ctx.Engines = append(ts.Ctx.Engines, engine.NewInMem(
+				roachpb.Attributes{},
+				storeSpec.SizeInBytes,
+				params.Stopper,
+			))
+		} else {
+			// TODO(bram): This will require both cleanup and on failure or if
+			// desired, not cleanup (perhaps on failure).
+			panic(fmt.Sprintf("test server does not yet support in non-memory stores: %s", storeSpec))
+		}
 	}
 
 	var err error

server/updates_test.go

@@ -95,7 +95,12 @@ func TestReportUsage(t *testing.T) {
 	}
 	defer stubURL(&reportingURL, u)()
 
-	params := base.TestServerArgs{StoresPerNode: 2}
+	params := base.TestServerArgs{
+		StoreSpecs: []base.StoreSpec{
+			base.DefaultTestStoreSpec,
+			base.DefaultTestStoreSpec,
+		},
+	}
 	s, _, _ := serverutils.StartServer(t, params)
 	ts := s.(*TestServer)
 
@@ -144,7 +149,7 @@ func TestReportUsage(t *testing.T) {
 	if minExpected, actual := totalRanges, reported.Node.RangeCount; minExpected > actual {
 		t.Errorf("expected node ranges at least %v got %v", minExpected, actual)
 	}
-	if minExpected, actual := params.StoresPerNode, len(reported.Stores); minExpected > actual {
+	if minExpected, actual := len(params.StoreSpecs), len(reported.Stores); minExpected > actual {
 		t.Errorf("expected at least %v stores got %v", minExpected, actual)
 	}
 

testutils/testcluster/testcluster.go

@@ -136,7 +136,13 @@ func StartTestCluster(t testing.TB, nodes int, args base.TestClusterArgs) *TestC
 
 	args.ServerArgs.PartOfCluster = true
 	for i := 0; i < nodes; i++ {
-		serverArgs := args.ServerArgs
+		var serverArgs base.TestServerArgs
+		if perNodeServerArgs, ok := args.ServerArgsPerNode[i]; ok {
+			serverArgs = perNodeServerArgs
+			serverArgs.PartOfCluster = true
+		} else {
+			serverArgs = args.ServerArgs
+		}
 		serverArgs.Stopper = stop.NewStopper()
 		if i > 0 {
 			serverArgs.JoinAddr = tc.Servers[0].ServingAddr()