1
0
Fork 0
You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

1342 lines
40 KiB
Go

package store
import (
"bytes"
"fmt"
"io/ioutil"
"net"
"os"
"path/filepath"
"sort"
"testing"
"time"
"github.com/rqlite/rqlite/command"
"github.com/rqlite/rqlite/command/encoding"
"github.com/rqlite/rqlite/testdata/chinook"
)
func Test_OpenStoreSingleNode(t *testing.T) {
s := mustNewStore(true)
defer os.RemoveAll(s.Path())
if err := s.Open(true); err != nil {
t.Fatalf("failed to open single-node store: %s", err.Error())
}
s.WaitForLeader(10 * time.Second)
got, err := s.LeaderAddr()
if err != nil {
t.Fatalf("failed to get leader address: %s", err.Error())
}
if exp := got; got != exp {
t.Fatalf("wrong leader address returned, got: %s, exp %s", got, exp)
}
id, err := s.LeaderID()
if err != nil {
t.Fatalf("failed to retrieve leader ID: %s", err.Error())
}
if got, exp := id, s.raftID; got != exp {
t.Fatalf("wrong leader ID returned, got: %s, exp %s", got, exp)
}
}
func Test_OpenStoreCloseSingleNode(t *testing.T) {
s := mustNewStore(true)
defer os.RemoveAll(s.Path())
if err := s.Open(true); err != nil {
t.Fatalf("failed to open single-node store: %s", err.Error())
}
s.WaitForLeader(10 * time.Second)
if err := s.Close(true); err != nil {
t.Fatalf("failed to close single-node store: %s", err.Error())
}
}
func Test_SingleNodeInMemExecuteQuery(t *testing.T) {
s := mustNewStore(true)
defer os.RemoveAll(s.Path())
if err := s.Open(true); err != nil {
t.Fatalf("failed to open single-node store: %s", err.Error())
}
defer s.Close(true)
s.WaitForLeader(10 * time.Second)
er := executeRequestFromStrings([]string{
`CREATE TABLE foo (id INTEGER NOT NULL PRIMARY KEY, name TEXT)`,
`INSERT INTO foo(id, name) VALUES(1, "fiona")`,
}, false, false)
_, err := s.Execute(er)
if err != nil {
t.Fatalf("failed to execute on single node: %s", err.Error())
}
qr := queryRequestFromString("SELECT * FROM foo", false, false)
qr.Level = command.QueryRequest_QUERY_REQUEST_LEVEL_NONE
r, err := s.Query(qr)
if err != nil {
t.Fatalf("failed to query single node: %s", err.Error())
}
if exp, got := `["id","name"]`, asJSON(r[0].Columns); exp != got {
t.Fatalf("unexpected results for query\nexp: %s\ngot: %s", exp, got)
}
if exp, got := `[[1,"fiona"]]`, asJSON(r[0].Values); exp != got {
t.Fatalf("unexpected results for query\nexp: %s\ngot: %s", exp, got)
}
}
// Test_SingleNodeInMemExecuteQueryFail ensures database level errors are presented by the store.
func Test_SingleNodeInMemExecuteQueryFail(t *testing.T) {
s := mustNewStore(true)
defer os.RemoveAll(s.Path())
if err := s.Open(true); err != nil {
t.Fatalf("failed to open single-node store: %s", err.Error())
}
defer s.Close(true)
s.WaitForLeader(10 * time.Second)
er := executeRequestFromStrings([]string{
`INSERT INTO foo(id, name) VALUES(1, "fiona")`,
}, false, false)
r, err := s.Execute(er)
if err != nil {
t.Fatalf("failed to execute on single node: %s", err.Error())
}
if exp, got := "no such table: foo", r[0].Error; exp != got {
t.Fatalf("unexpected results for query\nexp: %s\ngot: %s", exp, got)
}
}
func Test_SingleNodeFileExecuteQuery(t *testing.T) {
s := mustNewStore(false)
defer os.RemoveAll(s.Path())
if err := s.Open(true); err != nil {
t.Fatalf("failed to open single-node store: %s", err.Error())
}
defer s.Close(true)
s.WaitForLeader(10 * time.Second)
er := executeRequestFromStrings([]string{
`CREATE TABLE foo (id INTEGER NOT NULL PRIMARY KEY, name TEXT)`,
`INSERT INTO foo(id, name) VALUES(1, "fiona")`,
}, false, false)
_, err := s.Execute(er)
if err != nil {
t.Fatalf("failed to execute on single node: %s", err.Error())
}
// Every query should return the same results, so use a function for the check.
check := func(r []*command.QueryRows) {
if exp, got := `["id","name"]`, asJSON(r[0].Columns); exp != got {
t.Fatalf("unexpected results for query\nexp: %s\ngot: %s", exp, got)
}
if exp, got := `[[1,"fiona"]]`, asJSON(r[0].Values); exp != got {
t.Fatalf("unexpected results for query\nexp: %s\ngot: %s", exp, got)
}
}
qr := queryRequestFromString("SELECT * FROM foo", false, false)
qr.Level = command.QueryRequest_QUERY_REQUEST_LEVEL_NONE
r, err := s.Query(qr)
if err != nil {
t.Fatalf("failed to query single node: %s", err.Error())
}
check(r)
qr = queryRequestFromString("SELECT * FROM foo", false, false)
qr.Level = command.QueryRequest_QUERY_REQUEST_LEVEL_WEAK
r, err = s.Query(qr)
if err != nil {
t.Fatalf("failed to query single node: %s", err.Error())
}
check(r)
qr = queryRequestFromString("SELECT * FROM foo", false, false)
qr.Level = command.QueryRequest_QUERY_REQUEST_LEVEL_STRONG
r, err = s.Query(qr)
if err != nil {
t.Fatalf("failed to query single node: %s", err.Error())
}
check(r)
qr = queryRequestFromString("SELECT * FROM foo", false, true)
qr.Timings = true
qr.Level = command.QueryRequest_QUERY_REQUEST_LEVEL_NONE
r, err = s.Query(qr)
if err != nil {
t.Fatalf("failed to query single node: %s", err.Error())
}
check(r)
qr = queryRequestFromString("SELECT * FROM foo", true, false)
qr.Request.Transaction = true
qr.Level = command.QueryRequest_QUERY_REQUEST_LEVEL_NONE
r, err = s.Query(qr)
if err != nil {
t.Fatalf("failed to query single node: %s", err.Error())
}
check(r)
}
func Test_SingleNodeExecuteQueryTx(t *testing.T) {
s := mustNewStore(true)
defer os.RemoveAll(s.Path())
if err := s.Open(true); err != nil {
t.Fatalf("failed to open single-node store: %s", err.Error())
}
defer s.Close(true)
s.WaitForLeader(10 * time.Second)
er := executeRequestFromStrings([]string{
`CREATE TABLE foo (id INTEGER NOT NULL PRIMARY KEY, name TEXT)`,
`INSERT INTO foo(id, name) VALUES(1, "fiona")`,
}, false, true)
_, err := s.Execute(er)
if err != nil {
t.Fatalf("failed to execute on single node: %s", err.Error())
}
qr := queryRequestFromString("SELECT * FROM foo", false, true)
var r []*command.QueryRows
qr.Level = command.QueryRequest_QUERY_REQUEST_LEVEL_NONE
_, err = s.Query(qr)
if err != nil {
t.Fatalf("failed to query single node: %s", err.Error())
}
qr.Level = command.QueryRequest_QUERY_REQUEST_LEVEL_WEAK
_, err = s.Query(qr)
if err != nil {
t.Fatalf("failed to query single node: %s", err.Error())
}
qr.Level = command.QueryRequest_QUERY_REQUEST_LEVEL_STRONG
r, err = s.Query(qr)
if err != nil {
t.Fatalf("failed to query single node: %s", err.Error())
}
if exp, got := `["id","name"]`, asJSON(r[0].Columns); exp != got {
t.Fatalf("unexpected results for query\nexp: %s\ngot: %s", exp, got)
}
if exp, got := `[[1,"fiona"]]`, asJSON(r[0].Values); exp != got {
t.Fatalf("unexpected results for query\nexp: %s\ngot: %s", exp, got)
}
}
func Test_SingleNodeInMemFK(t *testing.T) {
s := mustNewStoreFK(true)
defer os.RemoveAll(s.Path())
if err := s.Open(true); err != nil {
t.Fatalf("failed to open single-node store: %s", err.Error())
}
defer s.Close(true)
s.WaitForLeader(10 * time.Second)
er := executeRequestFromStrings([]string{
`CREATE TABLE foo (id INTEGER NOT NULL PRIMARY KEY, name TEXT)`,
`CREATE TABLE bar (fooid INTEGER NOT NULL PRIMARY KEY, FOREIGN KEY(fooid) REFERENCES foo(id))`,
}, false, false)
_, err := s.Execute(er)
if err != nil {
t.Fatalf("failed to execute on single node: %s", err.Error())
}
res, err := s.Execute(executeRequestFromString("INSERT INTO bar(fooid) VALUES(1)", false, false))
if got, exp := asJSON(res), `[{"error":"FOREIGN KEY constraint failed"}]`; exp != got {
t.Fatalf("unexpected results for execute\nexp: %s\ngot: %s", exp, got)
}
}
func Test_SingleNodeBackupBinary(t *testing.T) {
t.Parallel()
s := mustNewStore(false)
defer os.RemoveAll(s.Path())
if err := s.Open(true); err != nil {
t.Fatalf("failed to open single-node store: %s", err.Error())
}
defer s.Close(true)
s.WaitForLeader(10 * time.Second)
dump := `PRAGMA foreign_keys=OFF;
BEGIN TRANSACTION;
CREATE TABLE foo (id integer not null primary key, name text);
INSERT INTO "foo" VALUES(1,'fiona');
COMMIT;
`
_, err := s.Execute(executeRequestFromString(dump, false, false))
if err != nil {
t.Fatalf("failed to load simple dump: %s", err.Error())
}
f, err := ioutil.TempFile("", "rqlite-baktest-")
defer os.Remove(f.Name())
s.logger.Printf("backup file is %s", f.Name())
if err := s.Backup(true, BackupBinary, f); err != nil {
t.Fatalf("Backup failed %s", err.Error())
}
// Check the backed up data by reading back up file, underlying SQLite file,
// and comparing the two.
bkp, err := ioutil.ReadFile(f.Name())
if err != nil {
t.Fatalf("Backup Failed: unable to read backup file, %s", err.Error())
}
dbFile, err := ioutil.ReadFile(filepath.Join(s.Path(), sqliteFile))
if err != nil {
t.Fatalf("Backup Failed: unable to read source SQLite file, %s", err.Error())
}
if ret := bytes.Compare(bkp, dbFile); ret != 0 {
t.Fatalf("Backup Failed: backup bytes are not same")
}
}
func Test_SingleNodeBackupText(t *testing.T) {
t.Parallel()
s := mustNewStore(true)
defer os.RemoveAll(s.Path())
if err := s.Open(true); err != nil {
t.Fatalf("failed to open single-node store: %s", err.Error())
}
defer s.Close(true)
s.WaitForLeader(10 * time.Second)
dump := `PRAGMA foreign_keys=OFF;
BEGIN TRANSACTION;
CREATE TABLE foo (id integer not null primary key, name text);
INSERT INTO "foo" VALUES(1,'fiona');
COMMIT;
`
_, err := s.Execute(executeRequestFromString(dump, false, false))
if err != nil {
t.Fatalf("failed to load simple dump: %s", err.Error())
}
f, err := ioutil.TempFile("", "rqlite-baktest-")
defer os.Remove(f.Name())
s.logger.Printf("backup file is %s", f.Name())
if err := s.Backup(true, BackupSQL, f); err != nil {
t.Fatalf("Backup failed %s", err.Error())
}
// Check the backed up data
bkp, err := ioutil.ReadFile(f.Name())
if err != nil {
t.Fatalf("Backup Failed: unable to read backup file, %s", err.Error())
}
if ret := bytes.Compare(bkp, []byte(dump)); ret != 0 {
t.Fatalf("Backup Failed: backup bytes are not same")
}
}
func Test_SingleNodeLoad(t *testing.T) {
s := mustNewStore(true)
defer os.RemoveAll(s.Path())
if err := s.Open(true); err != nil {
t.Fatalf("failed to open single-node store: %s", err.Error())
}
defer s.Close(true)
s.WaitForLeader(10 * time.Second)
dump := `PRAGMA foreign_keys=OFF;
BEGIN TRANSACTION;
CREATE TABLE foo (id integer not null primary key, name text);
INSERT INTO "foo" VALUES(1,'fiona');
COMMIT;
`
_, err := s.Execute(executeRequestFromString(dump, false, false))
if err != nil {
t.Fatalf("failed to load simple dump: %s", err.Error())
}
// Check that data were loaded correctly.
qr := queryRequestFromString("SELECT * FROM foo", false, true)
qr.Level = command.QueryRequest_QUERY_REQUEST_LEVEL_STRONG
r, err := s.Query(qr)
if err != nil {
t.Fatalf("failed to query single node: %s", err.Error())
}
if exp, got := `["id","name"]`, asJSON(r[0].Columns); exp != got {
t.Fatalf("unexpected results for query\nexp: %s\ngot: %s", exp, got)
}
if exp, got := `[[1,"fiona"]]`, asJSON(r[0].Values); exp != got {
t.Fatalf("unexpected results for query\nexp: %s\ngot: %s", exp, got)
}
}
func Test_SingleNodeSingleCommandTrigger(t *testing.T) {
s := mustNewStore(true)
defer os.RemoveAll(s.Path())
if err := s.Open(true); err != nil {
t.Fatalf("failed to open single-node store: %s", err.Error())
}
defer s.Close(true)
s.WaitForLeader(10 * time.Second)
dump := `PRAGMA foreign_keys=OFF;
BEGIN TRANSACTION;
CREATE TABLE foo (id integer primary key asc, name text);
INSERT INTO "foo" VALUES(1,'bob');
INSERT INTO "foo" VALUES(2,'alice');
INSERT INTO "foo" VALUES(3,'eve');
CREATE TABLE bar (nameid integer, age integer);
INSERT INTO "bar" VALUES(1,44);
INSERT INTO "bar" VALUES(2,46);
INSERT INTO "bar" VALUES(3,8);
CREATE VIEW foobar as select name as Person, Age as age from foo inner join bar on foo.id == bar.nameid;
CREATE TRIGGER new_foobar instead of insert on foobar begin insert into foo (name) values (new.Person); insert into bar (nameid, age) values ((select id from foo where name == new.Person), new.Age); end;
COMMIT;
`
_, err := s.Execute(executeRequestFromString(dump, false, false))
if err != nil {
t.Fatalf("failed to load dump with trigger: %s", err.Error())
}
// Check that the VIEW and TRIGGER are OK by using both.
er := executeRequestFromString("INSERT INTO foobar VALUES('jason', 16)", false, true)
r, err := s.Execute(er)
if err != nil {
t.Fatalf("failed to insert into view on single node: %s", err.Error())
}
if exp, got := int64(3), r[0].GetLastInsertId(); exp != got {
t.Fatalf("unexpected results for query\nexp: %d\ngot: %d", exp, got)
}
}
func Test_SingleNodeLoadNoStatements(t *testing.T) {
s := mustNewStore(true)
defer os.RemoveAll(s.Path())
if err := s.Open(true); err != nil {
t.Fatalf("failed to open single-node store: %s", err.Error())
}
defer s.Close(true)
s.WaitForLeader(10 * time.Second)
dump := `PRAGMA foreign_keys=OFF;
BEGIN TRANSACTION;
COMMIT;
`
_, err := s.Execute(executeRequestFromString(dump, false, false))
if err != nil {
t.Fatalf("failed to load dump with no commands: %s", err.Error())
}
}
func Test_SingleNodeLoadEmpty(t *testing.T) {
s := mustNewStore(true)
defer os.RemoveAll(s.Path())
if err := s.Open(true); err != nil {
t.Fatalf("failed to open single-node store: %s", err.Error())
}
defer s.Close(true)
s.WaitForLeader(10 * time.Second)
dump := ``
_, err := s.Execute(executeRequestFromString(dump, false, false))
if err != nil {
t.Fatalf("failed to load empty dump: %s", err.Error())
}
}
func Test_SingleNodeLoadChinook(t *testing.T) {
s := mustNewStore(true)
defer os.RemoveAll(s.Path())
if err := s.Open(true); err != nil {
t.Fatalf("failed to open single-node store: %s", err.Error())
}
defer s.Close(true)
s.WaitForLeader(10 * time.Second)
_, err := s.Execute(executeRequestFromString(chinook.DB, false, false))
if err != nil {
t.Fatalf("failed to load chinook dump: %s", err.Error())
}
// Check that data were loaded correctly.
qr := queryRequestFromString("SELECT count(*) FROM track", false, true)
qr.Level = command.QueryRequest_QUERY_REQUEST_LEVEL_STRONG
r, err := s.Query(qr)
if err != nil {
t.Fatalf("failed to query single node: %s", err.Error())
}
if exp, got := `["count(*)"]`, asJSON(r[0].Columns); exp != got {
t.Fatalf("unexpected results for query\nexp: %s\ngot: %s", exp, got)
}
if exp, got := `[[3503]]`, asJSON(r[0].Values); exp != got {
t.Fatalf("unexpected results for query\nexp: %s\ngot: %s", exp, got)
}
qr = queryRequestFromString("SELECT count(*) FROM album", false, true)
qr.Level = command.QueryRequest_QUERY_REQUEST_LEVEL_STRONG
r, err = s.Query(qr)
if err != nil {
t.Fatalf("failed to query single node: %s", err.Error())
}
if exp, got := `["count(*)"]`, asJSON(r[0].Columns); exp != got {
t.Fatalf("unexpected results for query\nexp: %s\ngot: %s", exp, got)
}
if exp, got := `[[347]]`, asJSON(r[0].Values); exp != got {
t.Fatalf("unexpected results for query\nexp: %s\ngot: %s", exp, got)
}
qr = queryRequestFromString("SELECT count(*) FROM artist", false, true)
qr.Level = command.QueryRequest_QUERY_REQUEST_LEVEL_STRONG
r, err = s.Query(qr)
if err != nil {
t.Fatalf("failed to query single node: %s", err.Error())
}
if exp, got := `["count(*)"]`, asJSON(r[0].Columns); exp != got {
t.Fatalf("unexpected results for query\nexp: %s\ngot: %s", exp, got)
}
if exp, got := `[[275]]`, asJSON(r[0].Values); exp != got {
t.Fatalf("unexpected results for query\nexp: %s\ngot: %s", exp, got)
}
}
func Test_MultiNodeJoinRemove(t *testing.T) {
s0 := mustNewStore(true)
defer os.RemoveAll(s0.Path())
if err := s0.Open(true); err != nil {
t.Fatalf("failed to open node for multi-node test: %s", err.Error())
}
defer s0.Close(true)
s0.WaitForLeader(10 * time.Second)
s1 := mustNewStore(true)
defer os.RemoveAll(s1.Path())
if err := s1.Open(false); err != nil {
t.Fatalf("failed to open node for multi-node test: %s", err.Error())
}
defer s1.Close(true)
// Get sorted list of cluster nodes.
storeNodes := []string{s0.ID(), s1.ID()}
sort.StringSlice(storeNodes).Sort()
// Join the second node to the first.
if err := s0.Join(s1.ID(), s1.Addr(), true); err != nil {
t.Fatalf("failed to join to node at %s: %s", s0.Addr(), err.Error())
}
s1.WaitForLeader(10 * time.Second)
got, err := s1.LeaderAddr()
if err != nil {
t.Fatalf("failed to get leader address: %s", err.Error())
}
// Check leader state on follower.
if exp := s0.Addr(); got != exp {
t.Fatalf("wrong leader address returned, got: %s, exp %s", got, exp)
}
id, err := s1.LeaderID()
if err != nil {
t.Fatalf("failed to retrieve leader ID: %s", err.Error())
}
if got, exp := id, s0.raftID; got != exp {
t.Fatalf("wrong leader ID returned, got: %s, exp %s", got, exp)
}
nodes, err := s0.Nodes()
if err != nil {
t.Fatalf("failed to get nodes: %s", err.Error())
}
if len(nodes) != len(storeNodes) {
t.Fatalf("size of cluster is not correct")
}
if storeNodes[0] != nodes[0].ID || storeNodes[1] != nodes[1].ID {
t.Fatalf("cluster does not have correct nodes")
}
// Remove a node.
if err := s0.Remove(s1.ID()); err != nil {
t.Fatalf("failed to remove %s from cluster: %s", s1.ID(), err.Error())
}
nodes, err = s0.Nodes()
if err != nil {
t.Fatalf("failed to get nodes post remove: %s", err.Error())
}
if len(nodes) != 1 {
t.Fatalf("size of cluster is not correct post remove")
}
if s0.ID() != nodes[0].ID {
t.Fatalf("cluster does not have correct nodes post remove")
}
}
func Test_MultiNodeJoinNonVoterRemove(t *testing.T) {
s0 := mustNewStore(true)
defer os.RemoveAll(s0.Path())
if err := s0.Open(true); err != nil {
t.Fatalf("failed to open node for multi-node test: %s", err.Error())
}
defer s0.Close(true)
s0.WaitForLeader(10 * time.Second)
s1 := mustNewStore(true)
defer os.RemoveAll(s1.Path())
if err := s1.Open(false); err != nil {
t.Fatalf("failed to open node for multi-node test: %s", err.Error())
}
defer s1.Close(true)
// Get sorted list of cluster nodes.
storeNodes := []string{s0.ID(), s1.ID()}
sort.StringSlice(storeNodes).Sort()
// Join the second node to the first.
if err := s0.Join(s1.ID(), s1.Addr(), false); err != nil {
t.Fatalf("failed to join to node at %s: %s", s0.Addr(), err.Error())
}
s1.WaitForLeader(10 * time.Second)
// Check leader state on follower.
got, err := s1.LeaderAddr()
if err != nil {
t.Fatalf("failed to get leader address: %s", err.Error())
}
if exp := s0.Addr(); got != exp {
t.Fatalf("wrong leader address returned, got: %s, exp %s", got, exp)
}
id, err := s1.LeaderID()
if err != nil {
t.Fatalf("failed to retrieve leader ID: %s", err.Error())
}
if got, exp := id, s0.raftID; got != exp {
t.Fatalf("wrong leader ID returned, got: %s, exp %s", got, exp)
}
nodes, err := s0.Nodes()
if err != nil {
t.Fatalf("failed to get nodes: %s", err.Error())
}
if len(nodes) != len(storeNodes) {
t.Fatalf("size of cluster is not correct")
}
if storeNodes[0] != nodes[0].ID || storeNodes[1] != nodes[1].ID {
t.Fatalf("cluster does not have correct nodes")
}
// Remove the non-voter.
if err := s0.Remove(s1.ID()); err != nil {
t.Fatalf("failed to remove %s from cluster: %s", s1.ID(), err.Error())
}
nodes, err = s0.Nodes()
if err != nil {
t.Fatalf("failed to get nodes post remove: %s", err.Error())
}
if len(nodes) != 1 {
t.Fatalf("size of cluster is not correct post remove")
}
if s0.ID() != nodes[0].ID {
t.Fatalf("cluster does not have correct nodes post remove")
}
}
func Test_MultiNodeExecuteQuery(t *testing.T) {
s0 := mustNewStore(true)
defer os.RemoveAll(s0.Path())
if err := s0.Open(true); err != nil {
t.Fatalf("failed to open node for multi-node test: %s", err.Error())
}
defer s0.Close(true)
s0.WaitForLeader(10 * time.Second)
s1 := mustNewStore(true)
defer os.RemoveAll(s1.Path())
if err := s1.Open(false); err != nil {
t.Fatalf("failed to open node for multi-node test: %s", err.Error())
}
defer s1.Close(true)
s2 := mustNewStore(true)
defer os.RemoveAll(s2.Path())
if err := s2.Open(false); err != nil {
t.Fatalf("failed to open node for multi-node test: %s", err.Error())
}
defer s2.Close(true)
// Join the second node to the first as a voting node.
if err := s0.Join(s1.ID(), s1.Addr(), true); err != nil {
t.Fatalf("failed to join to node at %s: %s", s0.Addr(), err.Error())
}
// Join the third node to the first as a non-voting node.
if err := s0.Join(s2.ID(), s2.Addr(), false); err != nil {
t.Fatalf("failed to join to node at %s: %s", s0.Addr(), err.Error())
}
er := executeRequestFromStrings([]string{
`CREATE TABLE foo (id INTEGER NOT NULL PRIMARY KEY, name TEXT)`,
`INSERT INTO foo(id, name) VALUES(1, "fiona")`,
}, false, false)
_, err := s0.Execute(er)
if err != nil {
t.Fatalf("failed to execute on single node: %s", err.Error())
}
s0FsmIdx, err := s0.WaitForAppliedFSM(5 * time.Second)
if err != nil {
t.Fatalf("failed to wait for fsmIndex: %s", err.Error())
}
qr := queryRequestFromString("SELECT * FROM foo", false, false)
qr.Level = command.QueryRequest_QUERY_REQUEST_LEVEL_NONE
r, err := s0.Query(qr)
if err != nil {
t.Fatalf("failed to query leader node: %s", err.Error())
}
if exp, got := `["id","name"]`, asJSON(r[0].Columns); exp != got {
t.Fatalf("unexpected results for query\nexp: %s\ngot: %s", exp, got)
}
if exp, got := `[[1,"fiona"]]`, asJSON(r[0].Values); exp != got {
t.Fatalf("unexpected results for query\nexp: %s\ngot: %s", exp, got)
}
// Wait until the log entries have been applied to the voting follower,
// and then query.
if _, err := s1.WaitForFSMIndex(s0FsmIdx, 5*time.Second); err != nil {
t.Fatalf("error waiting for follower to apply index: %s:", err.Error())
}
qr.Level = command.QueryRequest_QUERY_REQUEST_LEVEL_WEAK
_, err = s1.Query(qr)
if err == nil {
t.Fatalf("successfully queried non-leader node")
}
qr.Level = command.QueryRequest_QUERY_REQUEST_LEVEL_STRONG
_, err = s1.Query(qr)
if err == nil {
t.Fatalf("successfully queried non-leader node")
}
qr.Level = command.QueryRequest_QUERY_REQUEST_LEVEL_NONE
r, err = s1.Query(qr)
if err != nil {
t.Fatalf("failed to query follower node: %s", err.Error())
}
if exp, got := `["id","name"]`, asJSON(r[0].Columns); exp != got {
t.Fatalf("unexpected results for query\nexp: %s\ngot: %s", exp, got)
}
if exp, got := `[[1,"fiona"]]`, asJSON(r[0].Values); exp != got {
t.Fatalf("unexpected results for query\nexp: %s\ngot: %s", exp, got)
}
// Wait until the 3 log entries have been applied to the non-voting follower,
// and then query.
if err := s2.WaitForAppliedIndex(3, 5*time.Second); err != nil {
t.Fatalf("error waiting for follower to apply index: %s:", err.Error())
}
qr.Level = command.QueryRequest_QUERY_REQUEST_LEVEL_WEAK
_, err = s1.Query(qr)
if err == nil {
t.Fatalf("successfully queried non-voting node with Weak")
}
qr.Level = command.QueryRequest_QUERY_REQUEST_LEVEL_STRONG
_, err = s1.Query(qr)
if err == nil {
t.Fatalf("successfully queried non-voting node with Strong")
}
qr.Level = command.QueryRequest_QUERY_REQUEST_LEVEL_NONE
r, err = s1.Query(qr)
if err != nil {
t.Fatalf("failed to query non-voting node: %s", err.Error())
}
if exp, got := `["id","name"]`, asJSON(r[0].Columns); exp != got {
t.Fatalf("unexpected results for query\nexp: %s\ngot: %s", exp, got)
}
if exp, got := `[[1,"fiona"]]`, asJSON(r[0].Values); exp != got {
t.Fatalf("unexpected results for query\nexp: %s\ngot: %s", exp, got)
}
}
func Test_MultiNodeExecuteQueryFreshness(t *testing.T) {
s0 := mustNewStore(true)
defer os.RemoveAll(s0.Path())
if err := s0.Open(true); err != nil {
t.Fatalf("failed to open node for multi-node test: %s", err.Error())
}
defer s0.Close(true)
s0.WaitForLeader(10 * time.Second)
s1 := mustNewStore(true)
defer os.RemoveAll(s1.Path())
if err := s1.Open(false); err != nil {
t.Fatalf("failed to open node for multi-node test: %s", err.Error())
}
defer s1.Close(true)
// Join the second node to the first.
if err := s0.Join(s1.ID(), s1.Addr(), true); err != nil {
t.Fatalf("failed to join to node at %s: %s", s0.Addr(), err.Error())
}
er := executeRequestFromStrings([]string{
`CREATE TABLE foo (id INTEGER NOT NULL PRIMARY KEY, name TEXT)`,
`INSERT INTO foo(id, name) VALUES(1, "fiona")`,
}, false, false)
_, err := s0.Execute(er)
if err != nil {
t.Fatalf("failed to execute on single node: %s", err.Error())
}
qr := queryRequestFromString("SELECT * FROM foo", false, false)
qr.Level = command.QueryRequest_QUERY_REQUEST_LEVEL_NONE
r, err := s0.Query(qr)
if err != nil {
t.Fatalf("failed to query leader node: %s", err.Error())
}
if exp, got := `["id","name"]`, asJSON(r[0].Columns); exp != got {
t.Fatalf("unexpected results for query\nexp: %s\ngot: %s", exp, got)
}
if exp, got := `[[1,"fiona"]]`, asJSON(r[0].Values); exp != got {
t.Fatalf("unexpected results for query\nexp: %s\ngot: %s", exp, got)
}
// Wait until the 3 log entries have been applied to the follower,
// and then query.
if err := s1.WaitForAppliedIndex(3, 5*time.Second); err != nil {
t.Fatalf("error waiting for follower to apply index: %s:", err.Error())
}
// "Weak" consistency queries with 1 nanosecond freshness should pass, because freshness
// is ignored in this case.
qr.Level = command.QueryRequest_QUERY_REQUEST_LEVEL_WEAK
qr.Freshness = mustParseDuration("1ns").Nanoseconds()
_, err = s0.Query(qr)
if err != nil {
t.Fatalf("Failed to ignore freshness if level is Weak: %s", err.Error())
}
qr.Level = command.QueryRequest_QUERY_REQUEST_LEVEL_STRONG
// "Strong" consistency queries with 1 nanosecond freshness should pass, because freshness
// is ignored in this case.
_, err = s0.Query(qr)
if err != nil {
t.Fatalf("Failed to ignore freshness if level is Strong: %s", err.Error())
}
// Kill leader.
s0.Close(true)
// "None" consistency queries should still work.
qr.Level = command.QueryRequest_QUERY_REQUEST_LEVEL_NONE
qr.Freshness = 0
r, err = s1.Query(qr)
if err != nil {
t.Fatalf("failed to query follower node: %s", err.Error())
}
if exp, got := `["id","name"]`, asJSON(r[0].Columns); exp != got {
t.Fatalf("unexpected results for query\nexp: %s\ngot: %s", exp, got)
}
if exp, got := `[[1,"fiona"]]`, asJSON(r[0].Values); exp != got {
t.Fatalf("unexpected results for query\nexp: %s\ngot: %s", exp, got)
}
// Wait for the freshness interval to pass.
time.Sleep(mustParseDuration("1s"))
// "None" consistency queries with 1 nanosecond freshness should fail, because at least
// one nanosecond *should* have passed since leader died (surely!).
qr.Level = command.QueryRequest_QUERY_REQUEST_LEVEL_NONE
qr.Freshness = mustParseDuration("1ns").Nanoseconds()
_, err = s1.Query(qr)
if err == nil {
t.Fatalf("freshness violating query didn't return an error")
}
if err != ErrStaleRead {
t.Fatalf("freshness violating query didn't returned wrong error: %s", err.Error())
}
// Freshness of 0 is ignored.
qr.Freshness = 0
r, err = s1.Query(qr)
if err != nil {
t.Fatalf("failed to query follower node: %s", err.Error())
}
if exp, got := `["id","name"]`, asJSON(r[0].Columns); exp != got {
t.Fatalf("unexpected results for query\nexp: %s\ngot: %s", exp, got)
}
if exp, got := `[[1,"fiona"]]`, asJSON(r[0].Values); exp != got {
t.Fatalf("unexpected results for query\nexp: %s\ngot: %s", exp, got)
}
// "None" consistency queries with 1 hour freshness should pass, because it should
// not be that long since the leader died.
qr.Freshness = mustParseDuration("1h").Nanoseconds()
r, err = s1.Query(qr)
if err != nil {
t.Fatalf("failed to query follower node: %s", err.Error())
}
if exp, got := `["id","name"]`, asJSON(r[0].Columns); exp != got {
t.Fatalf("unexpected results for query\nexp: %s\ngot: %s", exp, got)
}
if exp, got := `[[1,"fiona"]]`, asJSON(r[0].Values); exp != got {
t.Fatalf("unexpected results for query\nexp: %s\ngot: %s", exp, got)
}
}
func Test_StoreLogTruncationMultinode(t *testing.T) {
s0 := mustNewStore(true)
defer os.RemoveAll(s0.Path())
s0.SnapshotThreshold = 4
s0.SnapshotInterval = 100 * time.Millisecond
if err := s0.Open(true); err != nil {
t.Fatalf("failed to open single-node store: %s", err.Error())
}
defer s0.Close(true)
s0.WaitForLeader(10 * time.Second)
nSnaps := stats.Get(numSnaphots).String()
// Write more than s.SnapshotThreshold statements.
queries := []string{
`CREATE TABLE foo (id INTEGER NOT NULL PRIMARY KEY, name TEXT)`,
`INSERT INTO foo(id, name) VALUES(1, "fiona")`,
`INSERT INTO foo(id, name) VALUES(2, "fiona")`,
`INSERT INTO foo(id, name) VALUES(3, "fiona")`,
`INSERT INTO foo(id, name) VALUES(4, "fiona")`,
`INSERT INTO foo(id, name) VALUES(5, "fiona")`,
}
for i := range queries {
_, err := s0.Execute(executeRequestFromString(queries[i], false, false))
if err != nil {
t.Fatalf("failed to execute on single node: %s", err.Error())
}
}
// Wait for the snapshot to happen and log to be truncated.
f := func() bool {
return stats.Get(numSnaphots).String() != nSnaps
}
testPoll(t, f, 100*time.Millisecond, 2*time.Second)
// Do one more execute, to ensure there is at least one log not snapshot.
// Without this, there is no guaratnee fsmIndex will be set on s1.
_, err := s0.Execute(executeRequestFromString(`INSERT INTO foo(id, name) VALUES(6, "fiona")`, false, false))
if err != nil {
t.Fatalf("failed to execute on single node: %s", err.Error())
}
// Fire up new node and ensure it picks up all changes. This will
// involve getting a snapshot and truncated log.
s1 := mustNewStore(true)
if err := s1.Open(true); err != nil {
t.Fatalf("failed to open single-node store: %s", err.Error())
}
defer s1.Close(true)
// Join the second node to the first.
if err := s0.Join(s1.ID(), s1.Addr(), true); err != nil {
t.Fatalf("failed to join to node at %s: %s", s0.Addr(), err.Error())
}
s1.WaitForLeader(10 * time.Second)
// Wait until the log entries have been applied to the follower,
// and then query.
if err := s1.WaitForAppliedIndex(8, 5*time.Second); err != nil {
t.Fatalf("error waiting for follower to apply index: %s:", err.Error())
}
qr := queryRequestFromString("SELECT count(*) FROM foo", false, true)
qr.Level = command.QueryRequest_QUERY_REQUEST_LEVEL_NONE
r, err := s1.Query(qr)
if err != nil {
t.Fatalf("failed to query single node: %s", err.Error())
}
if exp, got := `["count(*)"]`, asJSON(r[0].Columns); exp != got {
t.Fatalf("unexpected results for query\nexp: %s\ngot: %s", exp, got)
}
if exp, got := `[[6]]`, asJSON(r[0].Values); exp != got {
t.Fatalf("unexpected results for query\nexp: %s\ngot: %s", exp, got)
}
}
func Test_SingleNodeSnapshotOnDisk(t *testing.T) {
s := mustNewStore(false)
defer os.RemoveAll(s.Path())
if err := s.Open(true); err != nil {
t.Fatalf("failed to open single-node store: %s", err.Error())
}
defer s.Close(true)
s.WaitForLeader(10 * time.Second)
queries := []string{
`CREATE TABLE foo (id INTEGER NOT NULL PRIMARY KEY, name TEXT)`,
`INSERT INTO foo(id, name) VALUES(1, "fiona")`,
}
_, err := s.Execute(executeRequestFromStrings(queries, false, false))
if err != nil {
t.Fatalf("failed to execute on single node: %s", err.Error())
}
_, err = s.Query(queryRequestFromString("SELECT * FROM foo", false, false))
if err != nil {
t.Fatalf("failed to query single node: %s", err.Error())
}
// Snap the node and write to disk.
f, err := s.Snapshot()
if err != nil {
t.Fatalf("failed to snapshot node: %s", err.Error())
}
snapDir := mustTempDir()
defer os.RemoveAll(snapDir)
snapFile, err := os.Create(filepath.Join(snapDir, "snapshot"))
if err != nil {
t.Fatalf("failed to create snapshot file: %s", err.Error())
}
sink := &mockSnapshotSink{snapFile}
if err := f.Persist(sink); err != nil {
t.Fatalf("failed to persist snapshot to disk: %s", err.Error())
}
// Check restoration.
snapFile, err = os.Open(filepath.Join(snapDir, "snapshot"))
if err != nil {
t.Fatalf("failed to open snapshot file: %s", err.Error())
}
if err := s.Restore(snapFile); err != nil {
t.Fatalf("failed to restore snapshot from disk: %s", err.Error())
}
// Ensure database is back in the correct state.
r, err := s.Query(queryRequestFromString("SELECT * FROM foo", false, false))
if err != nil {
t.Fatalf("failed to query single node: %s", err.Error())
}
if exp, got := `["id","name"]`, asJSON(r[0].Columns); exp != got {
t.Fatalf("unexpected results for query\nexp: %s\ngot: %s", exp, got)
}
if exp, got := `[[1,"fiona"]]`, asJSON(r[0].Values); exp != got {
t.Fatalf("unexpected results for query\nexp: %s\ngot: %s", exp, got)
}
}
func Test_SingleNodeSnapshotInMem(t *testing.T) {
s := mustNewStore(true)
defer os.RemoveAll(s.Path())
if err := s.Open(true); err != nil {
t.Fatalf("failed to open single-node store: %s", err.Error())
}
defer s.Close(true)
s.WaitForLeader(10 * time.Second)
queries := []string{
`CREATE TABLE foo (id INTEGER NOT NULL PRIMARY KEY, name TEXT)`,
`INSERT INTO foo(id, name) VALUES(1, "fiona")`,
}
_, err := s.Execute(executeRequestFromStrings(queries, false, false))
if err != nil {
t.Fatalf("failed to execute on single node: %s", err.Error())
}
_, err = s.Query(queryRequestFromString("SELECT * FROM foo", false, false))
if err != nil {
t.Fatalf("failed to query single node: %s", err.Error())
}
// Snap the node and write to disk.
f, err := s.Snapshot()
if err != nil {
t.Fatalf("failed to snapshot node: %s", err.Error())
}
snapDir := mustTempDir()
defer os.RemoveAll(snapDir)
snapFile, err := os.Create(filepath.Join(snapDir, "snapshot"))
if err != nil {
t.Fatalf("failed to create snapshot file: %s", err.Error())
}
sink := &mockSnapshotSink{snapFile}
if err := f.Persist(sink); err != nil {
t.Fatalf("failed to persist snapshot to disk: %s", err.Error())
}
// Check restoration.
snapFile, err = os.Open(filepath.Join(snapDir, "snapshot"))
if err != nil {
t.Fatalf("failed to open snapshot file: %s", err.Error())
}
if err := s.Restore(snapFile); err != nil {
t.Fatalf("failed to restore snapshot from disk: %s", err.Error())
}
// Ensure database is back in the correct state.
r, err := s.Query(queryRequestFromString("SELECT * FROM foo", false, false))
if err != nil {
t.Fatalf("failed to query single node: %s", err.Error())
}
if exp, got := `["id","name"]`, asJSON(r[0].Columns); exp != got {
t.Fatalf("unexpected results for query\nexp: %s\ngot: %s", exp, got)
}
if exp, got := `[[1,"fiona"]]`, asJSON(r[0].Values); exp != got {
t.Fatalf("unexpected results for query\nexp: %s\ngot: %s", exp, got)
}
// Write a record, ensuring it works and can be queried back i.e. that
// the system remains functional.
_, err = s.Execute(executeRequestFromString(`INSERT INTO foo(id, name) VALUES(2, "fiona")`, false, false))
if err != nil {
t.Fatalf("failed to execute on single node: %s", err.Error())
}
r, err = s.Query(queryRequestFromString("SELECT COUNT(*) FROM foo", false, false))
if err != nil {
t.Fatalf("failed to query single node: %s", err.Error())
}
if exp, got := `["COUNT(*)"]`, asJSON(r[0].Columns); exp != got {
t.Fatalf("unexpected results for query\nexp: %s\ngot: %s", exp, got)
}
if exp, got := `[[2]]`, asJSON(r[0].Values); exp != got {
t.Fatalf("unexpected results for query\nexp: %s\ngot: %s", exp, got)
}
}
func Test_SingleNodeRestoreNoncompressed(t *testing.T) {
s := mustNewStore(false)
defer os.RemoveAll(s.Path())
if err := s.Open(true); err != nil {
t.Fatalf("failed to open single-node store: %s", err.Error())
}
defer s.Close(true)
s.WaitForLeader(10 * time.Second)
// Check restoration from a pre-compressed SQLite database snap.
// This is to test for backwards compatilibty of this code.
f, err := os.Open(filepath.Join("testdata", "noncompressed-sqlite-snap.bin"))
if err != nil {
t.Fatalf("failed to open snapshot file: %s", err.Error())
}
if err := s.Restore(f); err != nil {
t.Fatalf("failed to restore noncompressed snapshot from disk: %s", err.Error())
}
// Ensure database is back in the expected state.
r, err := s.Query(queryRequestFromString("SELECT count(*) FROM foo", false, false))
if err != nil {
t.Fatalf("failed to query single node: %s", err.Error())
}
if exp, got := `["count(*)"]`, asJSON(r[0].Columns); exp != got {
t.Fatalf("unexpected results for query\nexp: %s\ngot: %s", exp, got)
}
if exp, got := `[[5000]]`, asJSON(r[0].Values); exp != got {
t.Fatalf("unexpected results for query\nexp: %s\ngot: %s", exp, got)
}
}
func Test_SingleNodeNoop(t *testing.T) {
s0 := mustNewStore(true)
if err := s0.Open(true); err != nil {
t.Fatalf("failed to open single-node store: %s", err.Error())
}
defer s0.Close(true)
s0.WaitForLeader(10 * time.Second)
if err := s0.Noop("1"); err != nil {
t.Fatalf("failed to write noop command: %s", err.Error())
}
if s0.numNoops != 1 {
t.Fatalf("noop count is wrong, got: %d", s0.numNoops)
}
}
func Test_IsLeader(t *testing.T) {
s := mustNewStore(true)
defer os.RemoveAll(s.Path())
if err := s.Open(true); err != nil {
t.Fatalf("failed to open single-node store: %s", err.Error())
}
defer s.Close(true)
s.WaitForLeader(10 * time.Second)
if !s.IsLeader() {
t.Fatalf("single node is not leader!")
}
}
func Test_State(t *testing.T) {
s := mustNewStore(true)
defer os.RemoveAll(s.Path())
if err := s.Open(true); err != nil {
t.Fatalf("failed to open single-node store: %s", err.Error())
}
defer s.Close(true)
s.WaitForLeader(10 * time.Second)
state := s.State()
if state != Leader {
t.Fatalf("single node returned incorrect state (not Leader): %v", s)
}
}
func mustNewStoreAtPath(path string, inmem, fk bool) *Store {
cfg := NewDBConfig(inmem)
cfg.FKConstraints = fk
s := New(mustMockLister("localhost:0"), &StoreConfig{
DBConf: cfg,
Dir: path,
ID: path, // Could be any unique string.
})
if s == nil {
panic("failed to create new store")
}
return s
}
func mustNewStore(inmem bool) *Store {
return mustNewStoreAtPath(mustTempDir(), inmem, false)
}
func mustNewStoreFK(inmem bool) *Store {
return mustNewStoreAtPath(mustTempDir(), inmem, true)
}
type mockSnapshotSink struct {
*os.File
}
func (m *mockSnapshotSink) ID() string {
return "1"
}
func (m *mockSnapshotSink) Cancel() error {
return nil
}
type mockTransport struct {
ln net.Listener
}
type mockListener struct {
ln net.Listener
}
func mustMockLister(addr string) Listener {
ln, err := net.Listen("tcp", addr)
if err != nil {
panic("failed to create new listner")
}
return &mockListener{ln}
}
func (m *mockListener) Dial(addr string, timeout time.Duration) (net.Conn, error) {
return net.DialTimeout("tcp", addr, timeout)
}
func (m *mockListener) Accept() (net.Conn, error) { return m.ln.Accept() }
func (m *mockListener) Close() error { return m.ln.Close() }
func (m *mockListener) Addr() net.Addr { return m.ln.Addr() }
func mustTempDir() string {
var err error
path, err := ioutil.TempDir("", "rqlilte-test-")
if err != nil {
panic("failed to create temp dir")
}
return path
}
func mustParseDuration(t string) time.Duration {
d, err := time.ParseDuration(t)
if err != nil {
panic("failed to parse duration")
}
return d
}
func executeRequestFromString(s string, timings, tx bool) *command.ExecuteRequest {
return executeRequestFromStrings([]string{s}, timings, tx)
}
// queryRequestFromStrings converts a slice of strings into a command.ExecuteRequest
func executeRequestFromStrings(s []string, timings, tx bool) *command.ExecuteRequest {
stmts := make([]*command.Statement, len(s))
for i := range s {
stmts[i] = &command.Statement{
Sql: s[i],
}
}
return &command.ExecuteRequest{
Request: &command.Request{
Statements: stmts,
Transaction: tx,
},
Timings: timings,
}
}
func queryRequestFromString(s string, timings, tx bool) *command.QueryRequest {
return queryRequestFromStrings([]string{s}, timings, tx)
}
// queryRequestFromStrings converts a slice of strings into a command.QueryRequest
func queryRequestFromStrings(s []string, timings, tx bool) *command.QueryRequest {
stmts := make([]*command.Statement, len(s))
for i := range s {
stmts[i] = &command.Statement{
Sql: s[i],
}
}
return &command.QueryRequest{
Request: &command.Request{
Statements: stmts,
Transaction: tx,
},
Timings: timings,
}
}
func asJSON(v interface{}) string {
b, err := encoding.JSONMarshal(v)
if err != nil {
panic(fmt.Sprintf("failed to JSON marshal value: %s", err.Error()))
}
return string(b)
}
func testPoll(t *testing.T, f func() bool, p time.Duration, d time.Duration) {
tck := time.NewTicker(p)
defer tck.Stop()
tmr := time.NewTimer(d)
defer tmr.Stop()
for {
select {
case <-tck.C:
if f() {
return
}
case <-tmr.C:
t.Fatalf("timeout expired: %s", t.Name())
}
}
}