From 2c436545525ff3149b3c297d3c204d0c66514b02 Mon Sep 17 00:00:00 2001 From: Edd Robinson Date: Mon, 18 Sep 2017 17:50:13 +0100 Subject: [PATCH] Add some tests for concurrent server access --- tests/server_concurrent_test.go | 157 ++++++++++++++++++++++++++++++++ tests/server_helpers.go | 5 + 2 files changed, 162 insertions(+) create mode 100644 tests/server_concurrent_test.go diff --git a/tests/server_concurrent_test.go b/tests/server_concurrent_test.go new file mode 100644 index 0000000000..c40989a215 --- /dev/null +++ b/tests/server_concurrent_test.go @@ -0,0 +1,157 @@ +package tests + +import ( + "fmt" + "strings" + "sync" + "testing" + "time" + + "github.com/influxdata/influxdb/influxql" +) + +func TestConcurrentServer_WriteValues(t *testing.T) { + t.Parallel() + + if testing.Short() { + t.Skip("Skipping in short mode") + } + + s := OpenDefaultServer(NewConfig()) + defer s.Close() + + if _, ok := s.(*RemoteServer); ok { + t.Skip("Skipping. Not implemented on remote server") + } + + // The first %%d becomes a %d once fmt is done, so we can then inject new + // measurement names later on. + write := strings.Join([]string{ + fmt.Sprintf(`a%%[1]d val=23.2 %d`, mustParseTime(time.RFC3339Nano, "2000-01-01T00:00:00Z").UnixNano()), + fmt.Sprintf(`b%%[1]d val=23.2 %d`, mustParseTime(time.RFC3339Nano, "2000-01-01T00:00:00Z").UnixNano()), + fmt.Sprintf(`c%%[1]d val=23.2 %d`, mustParseTime(time.RFC3339Nano, "2000-01-01T00:00:00Z").UnixNano()), + fmt.Sprintf(`d%%[1]d val=23.2 %d`, mustParseTime(time.RFC3339Nano, "2000-01-01T00:00:00Z").UnixNano()), + }, "\n") + + var i int64 + var f1 = func() { + s.Write("db0", "rp0", fmt.Sprintf(write, i), nil) + } + + var f2 = func() { s.DropDatabase("db0") } + runTest(10*time.Second, f1, f2) +} + +func TestConcurrentServer_TagValues(t *testing.T) { + t.Parallel() + + if testing.Short() { + t.Skip("Skipping in short mode") + } + + s := OpenDefaultServer(NewConfig()) + defer s.Close() + + if _, ok := s.(*RemoteServer); ok { + t.Skip("Skipping. Not implemented on remote server") + } + + write := strings.Join([]string{ + fmt.Sprintf(`a,host=serverA,region=uswest val=23.2 %d`, mustParseTime(time.RFC3339Nano, "2000-01-01T00:00:00Z").UnixNano()), + fmt.Sprintf(`a,host=serverA,region=useast val=23.2 %d`, mustParseTime(time.RFC3339Nano, "2000-01-01T00:00:00Z").UnixNano()), + fmt.Sprintf(`a,host=serverA,region=ussouth val=23.2 %d`, mustParseTime(time.RFC3339Nano, "2000-01-01T00:00:00Z").UnixNano()), + fmt.Sprintf(`a,host=serverA,region=usnorth val=23.2 %d`, mustParseTime(time.RFC3339Nano, "2000-01-01T00:00:00Z").UnixNano()), + }, "\n") + + var f1 = func() { + s.Write("db0", "rp0", write, nil) + } + + stmt, err := influxql.ParseStatement(`SHOW TAG VALUES WITH KEY = "region"`) + if err != nil { + t.Fatal(err) + } + rewrite, err := influxql.RewriteStatement(stmt) + if err != nil { + t.Fatal(err) + } + + cond := rewrite.(*influxql.ShowTagValuesStatement).Condition + var f2 = func() { + srv, ok := s.(*LocalServer) + if !ok { + t.Fatal("Not a local server") + } + srv.TSDBStore.TagValues("db0", cond) + } + + var f3 = func() { s.DropDatabase("db0") } + runTest(10*time.Second, f1, f2, f3) +} + +func TestConcurrentServer_ShowMeasurements(t *testing.T) { + t.Parallel() + + if testing.Short() { + t.Skip("Skipping in short mode") + } + + s := OpenDefaultServer(NewConfig()) + defer s.Close() + + if _, ok := s.(*RemoteServer); ok { + t.Skip("Skipping. Not implemented on remote server") + } + + write := strings.Join([]string{ + fmt.Sprintf(`a,host=serverA,region=uswest val=23.2 %d`, mustParseTime(time.RFC3339Nano, "2000-01-01T00:00:00Z").UnixNano()), + fmt.Sprintf(`a,host=serverA,region=useast val=23.2 %d`, mustParseTime(time.RFC3339Nano, "2000-01-01T00:00:00Z").UnixNano()), + fmt.Sprintf(`a,host=serverA,region=ussouth val=23.2 %d`, mustParseTime(time.RFC3339Nano, "2000-01-01T00:00:00Z").UnixNano()), + fmt.Sprintf(`a,host=serverA,region=usnorth val=23.2 %d`, mustParseTime(time.RFC3339Nano, "2000-01-01T00:00:00Z").UnixNano()), + }, "\n") + + var f1 = func() { + s.Write("db0", "rp0", write, nil) + s.DropDatabase("db0") + } + + var f2 = func() { + srv, ok := s.(*LocalServer) + if !ok { + t.Fatal("Not a local server") + } + srv.TSDBStore.MeasurementNames("db0", nil) + } + + runTest(10*time.Second, f1, f2) +} + +// runTest continuously, and concurrently, runs the provided functions. No error +// checking is performed, runTest is simply trying to find panics. +func runTest(d time.Duration, fns ...func()) { + var wg sync.WaitGroup + done := make(chan struct{}) + go func() { + timer := time.NewTimer(d) + <-timer.C + close(done) + timer.Stop() + }() + + for _, fn := range fns { + wg.Add(1) + go func(f func()) { + defer wg.Done() + for { + select { + case <-done: + return + default: + f() + } + } + }(fn) + } + + wg.Wait() +} diff --git a/tests/server_helpers.go b/tests/server_helpers.go index a707109ae1..7931d7bd6f 100644 --- a/tests/server_helpers.go +++ b/tests/server_helpers.go @@ -36,6 +36,7 @@ type Server interface { CreateDatabase(db string) (*meta.DatabaseInfo, error) CreateDatabaseAndRetentionPolicy(db string, rp *meta.RetentionPolicySpec, makeDefault bool) error CreateSubscription(database, rp, name, mode string, destinations []string) error + DropDatabase(db string) error Reset() error Query(query string) (results string, err error) @@ -316,6 +317,10 @@ func (s *LocalServer) CreateSubscription(database, rp, name, mode string, destin func (s *LocalServer) DropDatabase(db string) error { s.mu.RLock() defer s.mu.RUnlock() + + if err := s.TSDBStore.DeleteDatabase(db); err != nil { + return err + } return s.MetaClient.DropDatabase(db) }