mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-08-28 02:44:59 +00:00
264 lines
7.3 KiB
Go
264 lines
7.3 KiB
Go
// Copyright 2022 Molecula Corp. (DBA FeatureBase).
|
|
// SPDX-License-Identifier: Apache-2.0
|
|
package pilosa
|
|
|
|
import (
|
|
"fmt"
|
|
"math/rand"
|
|
"reflect"
|
|
"testing"
|
|
"testing/quick"
|
|
"time"
|
|
|
|
"github.com/davecgh/go-spew/spew"
|
|
"github.com/featurebasedb/featurebase/v3/disco"
|
|
pnet "github.com/featurebasedb/featurebase/v3/net"
|
|
"github.com/featurebasedb/featurebase/v3/roaring"
|
|
"github.com/featurebasedb/featurebase/v3/testhook"
|
|
. "github.com/featurebasedb/featurebase/v3/vprint" // nolint:staticcheck
|
|
)
|
|
|
|
// newHolderWithTempPath returns a new instance of Holder.
|
|
func newHolderWithTempPath(tb testing.TB, backend string) *Holder {
|
|
path, err := testhook.TempDirInDir(tb, *TempDir, "pilosa-holder-")
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
cfg := mustHolderConfig()
|
|
cfg.StorageConfig.Backend = backend
|
|
h := NewHolder(path, cfg)
|
|
PanicOn(h.Open())
|
|
testhook.Cleanup(tb, func() {
|
|
h.Close()
|
|
})
|
|
return h
|
|
}
|
|
|
|
// newIndexWithTempPath returns a new instance of Index.
|
|
func newIndexWithTempPath(tb testing.TB, name string) *Index {
|
|
path, err := testhook.TempDirInDir(tb, *TempDir, "pilosa-index-")
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
cfg := DefaultHolderConfig()
|
|
cfg.StorageConfig.FsyncEnabled = false
|
|
cfg.RBFConfig.FsyncEnabled = false
|
|
h := NewHolder(path, cfg)
|
|
PanicOn(h.Open())
|
|
index, err := h.CreateIndex(name, IndexOptions{})
|
|
testhook.Cleanup(tb, func() {
|
|
h.Close()
|
|
})
|
|
if err != nil {
|
|
panic(err)
|
|
}
|
|
return index
|
|
}
|
|
|
|
// Ensure the cluster can fairly distribute partitions across the nodes.
|
|
func TestCluster_Owners(t *testing.T) {
|
|
c := cluster{
|
|
noder: disco.NewLocalNoder([]*disco.Node{
|
|
{URI: NewTestURIFromHostPort("serverA", 1000)},
|
|
{URI: NewTestURIFromHostPort("serverB", 1000)},
|
|
{URI: NewTestURIFromHostPort("serverC", 1000)},
|
|
}),
|
|
Hasher: NewTestModHasher(),
|
|
ReplicaN: 2,
|
|
}
|
|
|
|
cNodes := c.noder.Nodes()
|
|
|
|
// Create a snapshot of the cluster to use for node/partition calculations.
|
|
snap := c.NewSnapshot()
|
|
|
|
// Verify nodes are distributed.
|
|
if a := snap.PartitionNodes(0); !reflect.DeepEqual(a, []*disco.Node{cNodes[0], cNodes[1]}) {
|
|
t.Fatalf("unexpected owners: %s", spew.Sdump(a))
|
|
}
|
|
|
|
// Verify nodes go around the ring.
|
|
if a := snap.PartitionNodes(2); !reflect.DeepEqual(a, []*disco.Node{cNodes[2], cNodes[0]}) {
|
|
t.Fatalf("unexpected owners: %s", spew.Sdump(a))
|
|
}
|
|
}
|
|
|
|
// Ensure the partitioner can assign a fragment to a partition.
|
|
func TestCluster_Partition(t *testing.T) {
|
|
if err := quick.Check(func(index string, shard uint64, partitionN int) bool {
|
|
c := newCluster()
|
|
c.partitionN = partitionN
|
|
|
|
partitionID := disco.ShardToShardPartition(index, shard, partitionN)
|
|
if partitionID < 0 || partitionID >= partitionN {
|
|
t.Errorf("partition out of range: shard=%d, p=%d, n=%d", shard, partitionID, partitionN)
|
|
}
|
|
|
|
return true
|
|
}, &quick.Config{
|
|
Values: func(values []reflect.Value, rand *rand.Rand) {
|
|
values[0], _ = quick.Value(reflect.TypeOf(""), rand)
|
|
values[1] = reflect.ValueOf(uint64(rand.Uint32()))
|
|
values[2] = reflect.ValueOf(rand.Intn(1000) + 1)
|
|
},
|
|
}); err != nil {
|
|
t.Fatal(err)
|
|
}
|
|
}
|
|
|
|
// Ensure the hasher can hash correctly.
|
|
func TestHasher(t *testing.T) {
|
|
for _, tt := range []struct {
|
|
key uint64
|
|
bucket []int
|
|
}{
|
|
// Generated from the reference C++ code
|
|
{0, []int{0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0}},
|
|
{1, []int{0, 0, 0, 0, 0, 0, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 6, 17, 17}},
|
|
{0xdeadbeef, []int{0, 1, 2, 3, 3, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 5, 16, 16, 16}},
|
|
{0x0ddc0ffeebadf00d, []int{0, 1, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 2, 15, 15, 15, 15}},
|
|
} {
|
|
for i, v := range tt.bucket {
|
|
hasher := &disco.Jmphasher{}
|
|
if got := hasher.Hash(tt.key, i+1); got != v {
|
|
t.Errorf("hash(%v,%v)=%v, want %v", tt.key, i+1, got, v)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
// Ensure ContainsShards can find the actual shard list for node and index.
|
|
func TestCluster_ContainsShards(t *testing.T) {
|
|
c := NewTestCluster(t, 5)
|
|
c.ReplicaN = 3
|
|
cNodes := c.noder.Nodes()
|
|
|
|
// Create a snapshot of the cluster to use for node/partition calculations.
|
|
snap := c.NewSnapshot()
|
|
|
|
shards := snap.ContainsShards("test", roaring.NewBitmap(0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10), cNodes[2])
|
|
|
|
if !reflect.DeepEqual(shards, []uint64{0, 2, 3, 5, 6, 9, 10}) {
|
|
t.Fatalf("unexpected shars for node's index: %v", shards)
|
|
}
|
|
}
|
|
|
|
func TestCluster_Nodes(t *testing.T) {
|
|
const urisCount = 4
|
|
var uris []pnet.URI
|
|
arbitraryPorts := []int{17384, 17385, 17386, 17387}
|
|
for i := 0; i < urisCount; i++ {
|
|
uris = append(uris, NewTestURIFromHostPort(fmt.Sprintf("node%d", i), uint16(arbitraryPorts[i])))
|
|
}
|
|
|
|
node0 := &disco.Node{ID: "node0", URI: uris[0]}
|
|
node1 := &disco.Node{ID: "node1", URI: uris[1]}
|
|
node2 := &disco.Node{ID: "node2", URI: uris[2]}
|
|
node3 := &disco.Node{ID: "node3", URI: uris[3]}
|
|
|
|
nodes := []*disco.Node{node0, node1, node2}
|
|
|
|
t.Run("NodeIDs", func(t *testing.T) {
|
|
actual := disco.Nodes(nodes).IDs()
|
|
expected := []string{node0.ID, node1.ID, node2.ID}
|
|
if !reflect.DeepEqual(actual, expected) {
|
|
t.Errorf("expected: %v, but got: %v", expected, actual)
|
|
}
|
|
})
|
|
|
|
t.Run("Filter", func(t *testing.T) {
|
|
actual := disco.Nodes(disco.Nodes(nodes).Filter(nodes[1])).URIs()
|
|
expected := []pnet.URI{uris[0], uris[2]}
|
|
if !reflect.DeepEqual(actual, expected) {
|
|
t.Errorf("expected: %v, but got: %v", expected, actual)
|
|
}
|
|
})
|
|
|
|
t.Run("FilterURI", func(t *testing.T) {
|
|
actual := disco.Nodes(disco.Nodes(nodes).FilterURI(uris[1])).URIs()
|
|
expected := []pnet.URI{uris[0], uris[2]}
|
|
if !reflect.DeepEqual(actual, expected) {
|
|
t.Errorf("expected: %v, but got: %v", expected, actual)
|
|
}
|
|
})
|
|
|
|
t.Run("Contains", func(t *testing.T) {
|
|
actualTrue := disco.Nodes(nodes).Contains(node1)
|
|
actualFalse := disco.Nodes(nodes).Contains(node3)
|
|
if !reflect.DeepEqual(actualTrue, true) {
|
|
t.Errorf("expected: %v, but got: %v", true, actualTrue)
|
|
}
|
|
if !reflect.DeepEqual(actualFalse, false) {
|
|
t.Errorf("expected: %v, but got: %v", false, actualTrue)
|
|
}
|
|
})
|
|
|
|
t.Run("Clone", func(t *testing.T) {
|
|
clone := disco.Nodes(nodes).Clone()
|
|
actual := disco.Nodes(clone).URIs()
|
|
expected := []pnet.URI{uris[0], uris[1], uris[2]}
|
|
if !reflect.DeepEqual(actual, expected) {
|
|
t.Errorf("expected: %v, but got: %v", expected, actual)
|
|
}
|
|
})
|
|
}
|
|
|
|
func TestAE(t *testing.T) {
|
|
t.Run("AbortDoesn'tBlockUninitialized", func(t *testing.T) {
|
|
c := newCluster()
|
|
ch := make(chan struct{})
|
|
go func() {
|
|
c.abortAntiEntropy()
|
|
close(ch)
|
|
}()
|
|
defer c.abortAntiEntropyQ() // avoid leaking a goroutine.
|
|
select {
|
|
case <-ch:
|
|
return
|
|
case <-time.After(time.Second):
|
|
t.Fatalf("aborting anti entropy on a new cluster blocked")
|
|
}
|
|
})
|
|
|
|
t.Run("AbortBlocksInitialized", func(t *testing.T) {
|
|
c := newCluster()
|
|
c.initializeAntiEntropy()
|
|
|
|
ch := make(chan struct{})
|
|
go func() {
|
|
c.abortAntiEntropy()
|
|
close(ch)
|
|
}()
|
|
defer c.abortAntiEntropyQ() // avoid leak of goroutine.
|
|
select {
|
|
case <-ch:
|
|
t.Fatalf("aborting anti entropy on an initialized cluster didn't block")
|
|
case <-time.After(time.Microsecond * 100):
|
|
}
|
|
})
|
|
|
|
t.Run("AbortAntiEntropyQ", func(t *testing.T) {
|
|
c := newCluster()
|
|
c.initializeAntiEntropy()
|
|
if c.abortAntiEntropyQ() {
|
|
t.Fatalf("abortAntiEntropyQ should report false when abort not called")
|
|
}
|
|
go func() {
|
|
for {
|
|
if c.abortAntiEntropyQ() {
|
|
break
|
|
}
|
|
}
|
|
}()
|
|
ch := make(chan struct{})
|
|
go func() {
|
|
c.abortAntiEntropy()
|
|
close(ch)
|
|
}()
|
|
select {
|
|
case <-ch:
|
|
case <-time.After(time.Second):
|
|
t.Fatalf("abort should not have blocked this long")
|
|
}
|
|
})
|
|
}
|