Merge pull request #2057 from influxdb/fix_more_racy_tests

Move TestServer_RawDataReturnsInOrder to int tests

CHANGELOG.md

@@ -5,6 +5,7 @@
 - [#2039](https://github.com/influxdb/influxdb/pull/2039): Don't panic if getting current use fails
 - [#2034](https://github.com/influxdb/influxdb/pull/2034): Group by should require an aggregate.
 - [#2040](https://github.com/influxdb/influxdb/pull/2040): Add missing top-level help for config command
+- [#2057](https://github.com/influxdb/influxdb/pull/2057): Move racy "in order" test to integration test suite.
 
 ## v0.9.0-rc15 [2015-03-19]
 

cmd/influxd/server_integration_test.go

@@ -185,6 +185,7 @@ func query(t *testing.T, nodes Cluster, urlDb, query, expected string) (string,
 		v.Set("db", urlDb)
 	}
 
+	var actual string
 	// Query the data exists
 	for _, n := range nodes {
 		u := urlFor(n.url, "query", v)
@@ -198,13 +199,14 @@ func query(t *testing.T, nodes Cluster, urlDb, query, expected string) (string,
 		if err != nil {
 			t.Fatalf("Couldn't read body of response: %s", err.Error())
 		}
+		actual = string(body)
 
-		if expected != string(body) {
-			return string(body), false
+		if expected != actual {
+			return actual, false
 		}
 	}
 
-	return "", true
+	return actual, true
 }
 
 // queryAndWait executes the given query against all nodes in the cluster, and verifies no errors occured, and
@@ -269,6 +271,39 @@ var limitAndOffset = func(t *testing.T, node *Node, database, retention string)
 	}
 }
 
+func runTest_rawDataReturnsInOrder(t *testing.T, testName string, nodes Cluster, database, retention string) {
+	t.Logf("Running %s against %d-node cluster", testName, len(nodes))
+
+	// Start by ensuring database and retention policy exist.
+	createDatabase(t, testName, nodes, database)
+	createRetentionPolicy(t, testName, nodes, database, retention)
+	numPoints := 500
+	var expected string
+
+	for i := 1; i < numPoints; i++ {
+		data := fmt.Sprintf(`{"database": "%s", "retentionPolicy": "%s", "points": [{"name": "cpu", "timestamp": "%s", "tags": {"region": "us-east", "host": "server-%d"}, "fields": {"value": %d}}]}`,
+			database, retention, time.Unix(int64(i), int64(0)).Format(time.RFC3339), i%10, i)
+		write(t, nodes[0], data)
+	}
+
+	expected = fmt.Sprintf(`{"results":[{"series":[{"name":"cpu","columns":["time","count"],"values":[["1970-01-01T00:00:00Z",%d]]}]}]}`, numPoints-1)
+	_, ok := queryAndWait(t, nodes, database, `SELECT count(value) FROM cpu`, expected, 30*time.Second)
+	if !ok {
+		t.Errorf("test %s failed, SELECT count() query returned unexpected data", testName)
+	}
+
+	// Create expected JSON string dynamically.
+	expectedValues := make([]string, 0)
+	for i := 1; i < numPoints; i++ {
+		expectedValues = append(expectedValues, fmt.Sprintf(`["%s",%d]`, time.Unix(int64(i), int64(0)).UTC().Format(time.RFC3339), i))
+	}
+	expected = fmt.Sprintf(`{"results":[{"series":[{"name":"cpu","columns":["time","value"],"values":[%s]}]}]}`, strings.Join(expectedValues, ","))
+	_, ok = query(t, nodes, database, `SELECT value FROM cpu`, expected)
+	if !ok {
+		t.Errorf("test %s failed, SELECT query returned unexpected data", testName)
+	}
+}
+
 // runTests_Errors tests some basic error cases.
 func runTests_Errors(t *testing.T, nodes Cluster) {
 	t.Logf("Running tests against %d-node cluster", len(nodes))
@@ -1086,6 +1121,7 @@ func TestSingleServer(t *testing.T) {
 	nodes := createCombinedNodeCluster(t, testName, dir, 1, 8090, nil)
 
 	runTestsData(t, testName, nodes, "mydb", "myrp")
+	runTest_rawDataReturnsInOrder(t, testName, nodes, "mydb", "myrp")
 }
 
 func Test3NodeServer(t *testing.T) {
@@ -1103,6 +1139,8 @@ func Test3NodeServer(t *testing.T) {
 	nodes := createCombinedNodeCluster(t, testName, dir, 3, 8190, nil)
 
 	runTestsData(t, testName, nodes, "mydb", "myrp")
+	runTest_rawDataReturnsInOrder(t, testName, nodes, "mydb", "myrp")
+
 }
 
 func TestClientLibrary(t *testing.T) {

server_test.go

@@ -1377,53 +1377,6 @@ func TestServer_ShowSeriesLimitOffset(t *testing.T) {
 	}
 }
 
-// Ensure that when querying for raw data values that they return in time order
-func TestServer_RawDataReturnsInOrder(t *testing.T) {
-	c := test.NewMessagingClient()
-	defer c.Close()
-	s := OpenServer(c)
-	defer s.Close()
-	s.CreateDatabase("foo")
-	s.CreateRetentionPolicy("foo", &influxdb.RetentionPolicy{Name: "raw", Duration: 1 * time.Hour})
-	s.SetDefaultRetentionPolicy("foo", "raw")
-	s.CreateUser("susy", "pass", false)
-
-	for i := 1; i < 500; i++ {
-		host := fmt.Sprintf("server-%d", i%10)
-		s.MustWriteSeries("foo", "raw", []influxdb.Point{{Name: "cpu", Tags: map[string]string{"region": "us-east", "host": host}, Timestamp: time.Unix(int64(i), 0), Fields: map[string]interface{}{"value": float64(i)}}})
-	}
-
-	results := s.ExecuteQuery(MustParseQuery(`SELECT count(value) FROM cpu`), "foo", nil)
-	if res := results.Results[0]; res.Err != nil {
-		t.Fatalf("unexpected error during COUNT: %s", res.Err)
-	} else if s := mustMarshalJSON(res); s != `{"series":[{"name":"cpu","columns":["time","count"],"values":[["1970-01-01T00:00:00Z",499]]}]}` {
-		t.Fatalf("unexpected row(0) during COUNT: %s", s)
-	}
-
-	results = s.ExecuteQuery(MustParseQuery(`SELECT value FROM cpu`), "foo", nil)
-
-	lastTime := int64(0)
-	for _, v := range results.Results[0].Series[0].Values {
-		tt := v[0].(time.Time)
-		if lastTime > tt.UnixNano() {
-			t.Fatal("values out of order")
-		}
-		lastTime = tt.UnixNano()
-	}
-	if len(results.Results[0].Series[0].Values) != 499 {
-		t.Fatal("expected 499 values")
-	}
-
-	results = s.ExecuteQuery(MustParseQuery(`SELECT value FROM cpu GROUP BY *`), "foo", nil)
-	if res := results.Results[0]; res.Err != nil {
-		t.Fatalf("unexpected error during GROUP BY *: %s", res.Err)
-	} else if len(res.Series) != 10 {
-		t.Fatalf("expected 10 series back but got %d", len(res.Series))
-	} else if len(res.Series[1].Values) != 50 {
-		t.Fatalf("expected 50 values per series but got %d", len(res.Series[0].Values))
-	}
-}
-
 func TestServer_CreateShardGroupIfNotExist(t *testing.T) {
 	c := test.NewMessagingClient()
 	defer c.Close()