added load test for time and custom partition

* added load test for time and custom partition with batch and single events test * changes to smoke testing for custom partitions
parseablehq · Jun 5, 2024 · ea1d7df · ea1d7df
1 parent 6fe04df
commit ea1d7df
Show file tree

Hide file tree

Showing 2 changed files with 246 additions and 15 deletions.
diff --git a/quest_test.go b/quest_test.go
@@ -137,7 +137,7 @@ func TestLoadStreamBatchWithK6_StaticSchema(t *testing.T) {
 				"-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.QueryPassword),
 				"-e", fmt.Sprintf("P_STREAM=%s", staticSchemaStream),
 				"-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
-				"-e", fmt.Sprintf("P_EVENTS_COUNT=%s", schema_count),
+				"-e", fmt.Sprintf("P_EVENTS_COUNT=%s", events_count),
 				"./scripts/load_batch_events.js",
 				"--vus=", vus,
 				"--duration=", duration)
@@ -156,7 +156,7 @@ func TestLoadStreamBatchWithK6_StaticSchema(t *testing.T) {
 				"-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.IngestorPassword),
 				"-e", fmt.Sprintf("P_STREAM=%s", staticSchemaStream),
 				"-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
-				"-e", fmt.Sprintf("P_EVENTS_COUNT=%s", schema_count),
+				"-e", fmt.Sprintf("P_EVENTS_COUNT=%s", events_count),
 				"./scripts/load_batch_events.js",
 				"--vus=", vus,
 				"--duration=", duration)
@@ -289,13 +289,13 @@ func TestSmokeLoad_TimePartition_WithK6Stream(t *testing.T) {
 		cmd.Output()
 	}
 	time.Sleep(60 * time.Second)
-	QueryLogStreamCount_WithTimePartition(t, NewGlob.QueryClient, time_partition_stream, 60000)
+	QueryLogStreamCount(t, NewGlob.QueryClient, time_partition_stream, 60000)
 	DeleteStream(t, NewGlob.QueryClient, time_partition_stream)
 }
 
 func TestSmokeLoad_CustomPartition_WithK6Stream(t *testing.T) {
 	custom_partition_stream := NewGlob.Stream + "custompartition"
-	customHeader := map[string]string{"X-P-Custom-Partition": "level,os"}
+	customHeader := map[string]string{"X-P-Custom-Partition": "level"}
 	CreateStreamWithHeader(t, NewGlob.QueryClient, custom_partition_stream, customHeader)
 	if NewGlob.IngestorUrl.String() == "" {
 		cmd := exec.Command("k6",
@@ -327,7 +327,7 @@ func TestSmokeLoad_CustomPartition_WithK6Stream(t *testing.T) {
 
 func TestSmokeLoad_TimeAndCustomPartition_WithK6Stream(t *testing.T) {
 	custom_partition_stream := NewGlob.Stream + "timecustompartition"
-	customHeader := map[string]string{"X-P-Custom-Partition": "level,os", "X-P-Time-Partition": "source_time", "X-P-Time-Partition-Limit": "365d"}
+	customHeader := map[string]string{"X-P-Custom-Partition": "level", "X-P-Time-Partition": "source_time", "X-P-Time-Partition-Limit": "365d"}
 	CreateStreamWithHeader(t, NewGlob.QueryClient, custom_partition_stream, customHeader)
 	if NewGlob.IngestorUrl.String() == "" {
 		cmd := exec.Command("k6",
@@ -353,7 +353,7 @@ func TestSmokeLoad_TimeAndCustomPartition_WithK6Stream(t *testing.T) {
 		cmd.Output()
 	}
 	time.Sleep(60 * time.Second)
-	QueryLogStreamCount_WithTimePartition(t, NewGlob.QueryClient, custom_partition_stream, 60000)
+	QueryLogStreamCount(t, NewGlob.QueryClient, custom_partition_stream, 60000)
 	DeleteStream(t, NewGlob.QueryClient, custom_partition_stream)
 }
 
@@ -510,7 +510,7 @@ func TestLoadStreamBatchWithK6(t *testing.T) {
 				"-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.QueryPassword),
 				"-e", fmt.Sprintf("P_STREAM=%s", NewGlob.Stream),
 				"-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
-				"-e", fmt.Sprintf("P_EVENTS_COUNT=%s", schema_count),
+				"-e", fmt.Sprintf("P_EVENTS_COUNT=%s", events_count),
 				"./scripts/load_batch_events.js",
 				"--vus=", vus,
 				"--duration=", duration)
@@ -559,7 +559,7 @@ func TestLoadHistoricalStreamBatchWithK6(t *testing.T) {
 				"-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.QueryPassword),
 				"-e", fmt.Sprintf("P_STREAM=%s", historicalStream),
 				"-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
-				"-e", fmt.Sprintf("P_EVENTS_COUNT=%s", schema_count),
+				"-e", fmt.Sprintf("P_EVENTS_COUNT=%s", events_count),
 				"./scripts/load_historical_batch_events.js",
 				"--vus=", vus,
 				"--duration=", duration)
@@ -578,7 +578,7 @@ func TestLoadHistoricalStreamBatchWithK6(t *testing.T) {
 				"-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.IngestorPassword),
 				"-e", fmt.Sprintf("P_STREAM=%s", historicalStream),
 				"-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
-				"-e", fmt.Sprintf("P_EVENTS_COUNT=%s", schema_count),
+				"-e", fmt.Sprintf("P_EVENTS_COUNT=%s", events_count),
 				"./scripts/load_historical_batch_events.js",
 				"--vus=", vus,
 				"--duration=", duration)
@@ -595,6 +595,102 @@ func TestLoadHistoricalStreamBatchWithK6(t *testing.T) {
 	}
 }
 
+func TestLoadStreamBatchWithCustomPartitionWithK6(t *testing.T) {
+	customPartitionStream := NewGlob.Stream + "custompartition"
+	customHeader := map[string]string{"X-P-Custom-Partition": "level,os"}
+	CreateStreamWithHeader(t, NewGlob.QueryClient, customPartitionStream, customHeader)
+	if NewGlob.IngestorUrl.String() == "" {
+		cmd := exec.Command("k6",
+			"run",
+			"-e", fmt.Sprintf("P_URL=%s", NewGlob.QueryUrl.String()),
+			"-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.QueryUsername),
+			"-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.QueryPassword),
+			"-e", fmt.Sprintf("P_STREAM=%s", customPartitionStream),
+			"-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
+			"-e", fmt.Sprintf("P_EVENTS_COUNT=%s", events_count),
+			"./scripts/load_batch_events.js",
+			"--vus=", vus,
+			"--duration=", duration)
+
+		cmd.Run()
+		op, err := cmd.Output()
+		if err != nil {
+			t.Log(err)
+		}
+		t.Log(string(op))
+	} else {
+		cmd := exec.Command("k6",
+			"run",
+			"-e", fmt.Sprintf("P_URL=%s", NewGlob.IngestorUrl.String()),
+			"-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.IngestorUsername),
+			"-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.IngestorPassword),
+			"-e", fmt.Sprintf("P_STREAM=%s", customPartitionStream),
+			"-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
+			"-e", fmt.Sprintf("P_EVENTS_COUNT=%s", events_count),
+			"./scripts/load_batch_events.js",
+			"--vus=", vus,
+			"--duration=", duration)
+
+		cmd.Run()
+		op, err := cmd.Output()
+		if err != nil {
+			t.Log(err)
+		}
+		t.Log(string(op))
+	}
+
+	DeleteStream(t, NewGlob.QueryClient, customPartitionStream)
+}
+
+func TestLoadStreamBatchWithTimeAndCustomPartitionWithK6(t *testing.T) {
+	if NewGlob.Mode == "load" {
+		customPartitionStream := NewGlob.Stream + "timeandcustompartition"
+		customHeader := map[string]string{"X-P-Custom-Partition": "level,os", "X-P-Time-Partition": "source_time"}
+		CreateStreamWithHeader(t, NewGlob.QueryClient, customPartitionStream, customHeader)
+		if NewGlob.IngestorUrl.String() == "" {
+			cmd := exec.Command("k6",
+				"run",
+				"-e", fmt.Sprintf("P_URL=%s", NewGlob.QueryUrl.String()),
+				"-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.QueryUsername),
+				"-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.QueryPassword),
+				"-e", fmt.Sprintf("P_STREAM=%s", customPartitionStream),
+				"-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
+				"-e", fmt.Sprintf("P_EVENTS_COUNT=%s", events_count),
+				"./scripts/load_historical_batch_events.js",
+				"--vus=", vus,
+				"--duration=", duration)
+
+			cmd.Run()
+			op, err := cmd.Output()
+			if err != nil {
+				t.Log(err)
+			}
+			t.Log(string(op))
+		} else {
+			cmd := exec.Command("k6",
+				"run",
+				"-e", fmt.Sprintf("P_URL=%s", NewGlob.IngestorUrl.String()),
+				"-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.IngestorUsername),
+				"-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.IngestorPassword),
+				"-e", fmt.Sprintf("P_STREAM=%s", customPartitionStream),
+				"-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
+				"-e", fmt.Sprintf("P_EVENTS_COUNT=%s", events_count),
+				"./scripts/load_historical_batch_events.js",
+				"--vus=", vus,
+				"--duration=", duration)
+
+			cmd.Run()
+			op, err := cmd.Output()
+			if err != nil {
+				t.Log(err)
+			}
+			t.Log(string(op))
+		}
+
+		DeleteStream(t, NewGlob.QueryClient, customPartitionStream)
+	}
+}
+
 func TestLoadStreamNoBatchWithK6(t *testing.T) {
 	if NewGlob.Mode == "load" {
 		CreateStream(t, NewGlob.QueryClient, NewGlob.Stream)
@@ -639,6 +735,145 @@ func TestLoadStreamNoBatchWithK6(t *testing.T) {
 	}
 }
 
+func TestLoadHistoricalStreamNoBatchWithK6(t *testing.T) {
+	if NewGlob.Mode == "load" {
+		historicalStream := NewGlob.Stream + "historical"
+		timeHeader := map[string]string{"X-P-Time-Partition": "source_time"}
+		CreateStreamWithHeader(t, NewGlob.QueryClient, historicalStream, timeHeader)
+		if NewGlob.IngestorUrl.String() == "" {
+			cmd := exec.Command("k6",
+				"run",
+				"-e", fmt.Sprintf("P_URL=%s", NewGlob.QueryUrl.String()),
+				"-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.QueryUsername),
+				"-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.QueryPassword),
+				"-e", fmt.Sprintf("P_STREAM=%s", historicalStream),
+				"-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
+				"./scripts/load_single_events.js",
+				"--vus=", vus,
+				"--duration=", duration)
+
+			cmd.Run()
+			op, err := cmd.Output()
+			if err != nil {
+				t.Log(err)
+			}
+			t.Log(string(op))
+		} else {
+			cmd := exec.Command("k6",
+				"run",
+				"-e", fmt.Sprintf("P_URL=%s", NewGlob.IngestorUrl.String()),
+				"-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.IngestorUsername),
+				"-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.IngestorPassword),
+				"-e", fmt.Sprintf("P_STREAM=%s", historicalStream),
+				"-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
+				"./scripts/load_single_events.js",
+				"--vus=", vus,
+				"--duration=", duration)
+
+			cmd.Run()
+			op, err := cmd.Output()
+			if err != nil {
+				t.Log(err)
+			}
+			t.Log(string(op))
+		}
+
+		DeleteStream(t, NewGlob.QueryClient, historicalStream)
+	}
+}
+
+func TestLoadStreamNoBatchWithCustomPartitionWithK6(t *testing.T) {
+	customPartitionStream := NewGlob.Stream + "custompartition"
+	customHeader := map[string]string{"X-P-Custom-Partition": "level,os"}
+	CreateStreamWithHeader(t, NewGlob.QueryClient, customPartitionStream, customHeader)
+	if NewGlob.IngestorUrl.String() == "" {
+		cmd := exec.Command("k6",
+			"run",
+			"-e", fmt.Sprintf("P_URL=%s", NewGlob.QueryUrl.String()),
+			"-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.QueryUsername),
+			"-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.QueryPassword),
+			"-e", fmt.Sprintf("P_STREAM=%s", customPartitionStream),
+			"-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
+			"./scripts/load_single_events.js",
+			"--vus=", vus,
+			"--duration=", duration)
+
+		cmd.Run()
+		op, err := cmd.Output()
+		if err != nil {
+			t.Log(err)
+		}
+		t.Log(string(op))
+	} else {
+		cmd := exec.Command("k6",
+			"run",
+			"-e", fmt.Sprintf("P_URL=%s", NewGlob.IngestorUrl.String()),
+			"-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.IngestorUsername),
+			"-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.IngestorPassword),
+			"-e", fmt.Sprintf("P_STREAM=%s", customPartitionStream),
+			"-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
+			"./scripts/load_single_events.js",
+			"--vus=", vus,
+			"--duration=", duration)
+
+		cmd.Run()
+		op, err := cmd.Output()
+		if err != nil {
+			t.Log(err)
+		}
+		t.Log(string(op))
+	}
+
+	DeleteStream(t, NewGlob.QueryClient, customPartitionStream)
+}
+
+func TestLoadStreamNoBatchWithTimeAndCustomPartitionWithK6(t *testing.T) {
+	if NewGlob.Mode == "load" {
+		customPartitionStream := NewGlob.Stream + "timeandcustompartition"
+		customHeader := map[string]string{"X-P-Custom-Partition": "level,os", "X-P-Time-Partition": "source_time"}
+		CreateStreamWithHeader(t, NewGlob.QueryClient, customPartitionStream, customHeader)
+		if NewGlob.IngestorUrl.String() == "" {
+			cmd := exec.Command("k6",
+				"run",
+				"-e", fmt.Sprintf("P_URL=%s", NewGlob.QueryUrl.String()),
+				"-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.QueryUsername),
+				"-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.QueryPassword),
+				"-e", fmt.Sprintf("P_STREAM=%s", customPartitionStream),
+				"-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
+				"./scripts/load_single_events.js",
+				"--vus=", vus,
+				"--duration=", duration)
+
+			cmd.Run()
+			op, err := cmd.Output()
+			if err != nil {
+				t.Log(err)
+			}
+			t.Log(string(op))
+		} else {
+			cmd := exec.Command("k6",
+				"run",
+				"-e", fmt.Sprintf("P_URL=%s", NewGlob.IngestorUrl.String()),
+				"-e", fmt.Sprintf("P_USERNAME=%s", NewGlob.IngestorUsername),
+				"-e", fmt.Sprintf("P_PASSWORD=%s", NewGlob.IngestorPassword),
+				"-e", fmt.Sprintf("P_STREAM=%s", customPartitionStream),
+				"-e", fmt.Sprintf("P_SCHEMA_COUNT=%s", schema_count),
+				"./scripts/load_single_events.js",
+				"--vus=", vus,
+				"--duration=", duration)
+
+			cmd.Run()
+			op, err := cmd.Output()
+			if err != nil {
+				t.Log(err)
+			}
+			t.Log(string(op))
+		}
+
+		DeleteStream(t, NewGlob.QueryClient, customPartitionStream)
+	}
+}
+
 func TestDeleteStream(t *testing.T) {
 	DeleteStream(t, NewGlob.QueryClient, NewGlob.Stream)
 }
diff --git a/scripts/smoke.js b/scripts/smoke.js
@@ -16,12 +16,8 @@ export const options = {
 };
 
 function current_time() {
-    const startDate = new Date('2024-05-17T15:00:00');
-    const endDate = new Date('2024-05-17T15:03:59');
-    const timeDiff = endDate.getTime() - startDate.getTime();
-    const randomTime = Math.random() * timeDiff;
-    const randomDate = new Date(startDate.getTime() + randomTime);
-    return(randomDate.toISOString());
+    let event = new Date();
+    return event.toISOString();
 }
 
 function schemas() {