added test to verify count for historical ingestions (#65)

reduced iteration count in smoke test to 2000
reduced schema count to 10 in load test
reduced duration to 2 mins in load test
reduced event count to 5 in load test

quest_test.go

@@ -29,9 +29,9 @@ import (
 
 const (
 	vus          = "10"
-	duration     = "5m"
-	schema_count = "20"
-	events_count = "10"
+	duration     = "2m"
+	schema_count = "10"
+	events_count = "5"
 )
 
 func TestSmokeListLogStream(t *testing.T) {
@@ -256,7 +256,7 @@ func TestSmokeLoadWithK6Stream(t *testing.T) {
 		cmd.Output()
 	}
 	time.Sleep(120 * time.Second)
-	QueryLogStreamCount(t, NewGlob.QueryClient, NewGlob.Stream, 60000)
+	QueryLogStreamCount(t, NewGlob.QueryClient, NewGlob.Stream, 20000)
 	AssertStreamSchema(t, NewGlob.QueryClient, NewGlob.Stream, SchemaBody)
 	DeleteStream(t, NewGlob.QueryClient, NewGlob.Stream)
 }
@@ -289,7 +289,7 @@ func TestSmokeLoad_TimePartition_WithK6Stream(t *testing.T) {
 		cmd.Output()
 	}
 	time.Sleep(60 * time.Second)
-	QueryLogStreamCount(t, NewGlob.QueryClient, time_partition_stream, 60000)
+	QueryLogStreamCount_Historical(t, NewGlob.QueryClient, time_partition_stream, 20000)
 	DeleteStream(t, NewGlob.QueryClient, time_partition_stream)
 }
 
@@ -321,7 +321,7 @@ func TestSmokeLoad_CustomPartition_WithK6Stream(t *testing.T) {
 		cmd.Output()
 	}
 	time.Sleep(120 * time.Second)
-	QueryLogStreamCount(t, NewGlob.QueryClient, custom_partition_stream, 60000)
+	QueryLogStreamCount(t, NewGlob.QueryClient, custom_partition_stream, 20000)
 	DeleteStream(t, NewGlob.QueryClient, custom_partition_stream)
 }
 
@@ -353,7 +353,7 @@ func TestSmokeLoad_TimeAndCustomPartition_WithK6Stream(t *testing.T) {
 		cmd.Output()
 	}
 	time.Sleep(120 * time.Second)
-	QueryLogStreamCount(t, NewGlob.QueryClient, custom_partition_stream, 60000)
+	QueryLogStreamCount_Historical(t, NewGlob.QueryClient, custom_partition_stream, 20000)
 	DeleteStream(t, NewGlob.QueryClient, custom_partition_stream)
 }
 

scripts/load_historical_batch_events.js

@@ -17,7 +17,8 @@ export const options = {
 
 function current_time() {
     let event = new Date();
-    return event.toUTCString();
+    event.setMonth(event.getMonth() - 1);
+    return event.toISOString();
 }
 
 function schemas() {

scripts/smoke.js

@@ -10,13 +10,14 @@ export const options = {
         contacts: {
             executor: 'shared-iterations',
             vus: 20,
-            iterations: 6000,
+            iterations: 2000,
         },
     },
 };
 
 function current_time() {
     let event = new Date();
+    event.setMonth(event.getMonth() - 1);
     return event.toISOString();
 }
 

test_utils.go

@@ -265,6 +265,27 @@ func QueryLogStreamCount(t *testing.T, client HTTPClient, stream string, count u
 	require.Equalf(t, expected, body, "Query count incorrect; Expected %s, Actual %s", expected, body)
 }
 
+func QueryLogStreamCount_Historical(t *testing.T, client HTTPClient, stream string, count uint64) {
+	// Query last 30 minutes of data only
+	now := time.Now()
+	startTime := now.AddDate(0, 0, -32).Format(time.RFC3339Nano)
+	endTime := now.AddDate(0, 0, -30).Format(time.RFC3339Nano)
+
+	query := map[string]interface{}{
+		"query":     "select count(*) as count from " + stream,
+		"startTime": startTime,
+		"endTime":   endTime,
+	}
+	queryJSON, _ := json.Marshal(query)
+	req, _ := client.NewRequest("POST", "query", bytes.NewBuffer(queryJSON))
+	response, err := client.Do(req)
+	require.NoErrorf(t, err, "Request failed: %s", err)
+	body := readAsString(response.Body)
+	require.Equalf(t, 200, response.StatusCode, "Server returned http code: %s and response: %s", response.Status, body)
+	expected := fmt.Sprintf(`[{"count":%d}]`, count)
+	require.Equalf(t, expected, body, "Query count incorrect; Expected %s, Actual %s", expected, body)
+}
+
 func QueryTwoLogStreamCount(t *testing.T, client HTTPClient, stream1 string, stream2 string, count uint64) {
 	// Query last 30 minutes of data only
 	endTime := time.Now().Add(time.Second).Format(time.RFC3339Nano)