Skip to content

Commit 3af5d9e

Browse files
mihkelsmihkels
committed
test: Added Pravega e2e tests, ref: NOISSUE
1 parent 4335573 commit 3af5d9e

File tree

5 files changed

+265
-9
lines changed

5 files changed

+265
-9
lines changed

‎driver-bookkeeper/src/main/java/io/openmessaging/benchmark/driver/bookkeeper/DlogBenchmarkDriver.java‎

Lines changed: 3 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -51,13 +51,13 @@ public class DlogBenchmarkDriver implements BenchmarkDriver {
5151

5252
@Override
5353
public void initialize(File configurationFile, PrometheusMeterRegistry prometheusRegistry)
54-
throws IOException, InterruptedException {
54+
throws IOException {
5555
config = mapper.readValue(configurationFile, Config.class);
5656

5757
DistributedLogConfiguration conf = new DistributedLogConfiguration();
5858
try {
59-
PropertiesConfiguration propsConf =
60-
new Configurations().properties(new StringReader(config.dlogConf).toString());
59+
PropertiesConfiguration propsConf = new PropertiesConfiguration();
60+
propsConf.read(new StringReader(config.dlogConf));
6161
conf.loadConf(propsConf);
6262
} catch (ConfigurationException e) {
6363
log.error("Failed to load dlog configuration : \n{}\n", config.dlogConf, e);

‎e2e-tests/pom.xml‎

Lines changed: 18 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -57,6 +57,24 @@
5757
<version>${project.version}</version>
5858
<scope>test</scope>
5959
</dependency>
60+
<dependency>
61+
<groupId>${project.groupId}</groupId>
62+
<artifactId>driver-pravega</artifactId>
63+
<version>${project.version}</version>
64+
<scope>test</scope>
65+
</dependency>
66+
<dependency>
67+
<groupId>org.apache.curator</groupId>
68+
<artifactId>curator-framework</artifactId>
69+
<version>5.5.0</version>
70+
<scope>test</scope>
71+
</dependency>
72+
<dependency>
73+
<groupId>io.openmessaging.benchmark</groupId>
74+
<artifactId>driver-bookkeeper</artifactId>
75+
<version>${project.version}</version>
76+
<scope>test</scope>
77+
</dependency>
6078
<dependency>
6179
<groupId>org.apache.logging.log4j</groupId>
6280
<artifactId>log4j-slf4j2-impl</artifactId>

‎e2e-tests/src/test/java/io/openmessaging/benchmark/e2e/BaseE2eIT.java‎

Lines changed: 27 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -124,18 +124,39 @@ protected File createWorkloadFile(Workload workload) throws IOException {
124124
*/
125125
protected void validateResults(TestResult result, Workload workload) {
126126
var wholeTestDuration = workload.testDurationMinutes + workload.warmupDurationMinutes;
127-
log.info("Test completed - Aggregate throughput: {} msg/s, {} Mbit/s",
128-
result.aggregatedPublishLatencyAvg / wholeTestDuration,
129-
result.aggregatedPublishLatencyAvg * result.messageSize * 8.0 / 1024 / 1024 / wholeTestDuration);
130127

131-
// Basic validations
132-
assertThat(result.aggregatedPublishLatencyAvg)
128+
// Calculate actual aggregate rates
129+
double aggregatePublishRate = result.publishRate.stream()
130+
.mapToDouble(Double::doubleValue)
131+
.sum() / result.publishRate.size();
132+
133+
double aggregateConsumeRate = result.consumeRate.stream()
134+
.mapToDouble(Double::doubleValue)
135+
.sum() / result.consumeRate.size();
136+
137+
log.info("Test completed - Aggregate publish rate: {} msg/s, consume rate: {} msg/s, throughput: {} Mbit/s",
138+
aggregatePublishRate,
139+
aggregateConsumeRate,
140+
aggregatePublishRate * workload.messageSize * 8.0 / 1024 / 1024);
141+
142+
// Validate that messages were published
143+
assertThat(aggregatePublishRate)
133144
.as("Should have published messages")
134145
.isGreaterThan(0);
135146

136-
assertThat(result.aggregatedEndToEndLatencyAvg)
147+
// Validate that messages were consumed
148+
assertThat(aggregateConsumeRate)
137149
.as("Should have consumed messages")
138150
.isGreaterThan(0);
151+
152+
// Optional: Validate latencies are reasonable
153+
assertThat(result.aggregatedPublishLatencyAvg)
154+
.as("Publish latency should be positive")
155+
.isGreaterThan(0);
156+
157+
// assertThat(result.aggregatedEndToEndLatencyAvg)
158+
// .as("End-to-end latency should be positive")
159+
// .isGreaterThan(0);
139160
}
140161

141162
private void deleteDirectory(File directory) {

‎e2e-tests/src/test/java/io/openmessaging/benchmark/e2e/PravegaContainer.java‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,37 @@
1+
package io.openmessaging.benchmark.e2e;
2+
3+
import org.testcontainers.containers.GenericContainer;
4+
import org.testcontainers.utility.DockerImageName;
5+
6+
public class PravegaContainer extends GenericContainer<PravegaContainer> {
7+
8+
private static final int CONTROLLER_PORT = 9090;
9+
private static final int SEGMENT_STORE_PORT = 12345;
10+
private static final String DEFAULT_IMAGE = "pravega/pravega:0.13.0";
11+
private static final String HOST_IP = "127.0.0.1";
12+
13+
public PravegaContainer() {
14+
this(DEFAULT_IMAGE);
15+
}
16+
17+
public PravegaContainer(String imageName) {
18+
super(DockerImageName.parse(imageName));
19+
20+
withCommand("standalone");
21+
withEnv("JAVA_OPTS", "-Xmx1g");
22+
withEnv("HOST_IP", HOST_IP);
23+
24+
// Use the protected method in our subclass
25+
addFixedExposedPort(CONTROLLER_PORT, CONTROLLER_PORT);
26+
addFixedExposedPort(SEGMENT_STORE_PORT, SEGMENT_STORE_PORT);
27+
}
28+
29+
public String getControllerURI() {
30+
return "tcp://" + HOST_IP + ":" + CONTROLLER_PORT;
31+
}
32+
33+
public String getSegmentStoreEndpoint() {
34+
return HOST_IP + ":" + SEGMENT_STORE_PORT;
35+
}
36+
}
37+

‎e2e-tests/src/test/java/io/openmessaging/benchmark/e2e/PravegaE2eIT.java‎

Lines changed: 180 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,180 @@
1+
package io.openmessaging.benchmark.e2e;
2+
3+
import io.openmessaging.benchmark.TestResult;
4+
import io.openmessaging.benchmark.Workload;
5+
import io.openmessaging.benchmark.WorkloadGenerator;
6+
import io.openmessaging.benchmark.worker.LocalWorker;
7+
import io.pravega.client.ClientConfig;
8+
import io.pravega.client.admin.StreamManager;
9+
import org.junit.jupiter.api.AfterAll;
10+
import org.junit.jupiter.api.BeforeAll;
11+
import org.junit.jupiter.api.Test;
12+
import org.slf4j.Logger;
13+
import org.slf4j.LoggerFactory;
14+
import org.testcontainers.containers.wait.strategy.Wait;
15+
import org.testcontainers.junit.jupiter.Container;
16+
import org.testcontainers.junit.jupiter.Testcontainers;
17+
18+
import java.io.File;
19+
import java.net.URI;
20+
import java.nio.file.Files;
21+
import java.nio.file.Path;
22+
import java.time.Duration;
23+
24+
import static org.assertj.core.api.Assertions.assertThat;
25+
26+
/**
27+
* End-to-end tests for Pravega using a standalone container.
28+
*/
29+
@Testcontainers
30+
class PravegaE2eIT extends BaseE2eIT {
31+
32+
private static final Logger log = LoggerFactory.getLogger(PravegaE2eIT.class);
33+
34+
private static final String SCOPE = "ombscope";
35+
36+
@Container
37+
static PravegaContainer pravega = new PravegaContainer()
38+
.waitingFor(
39+
Wait.forLogMessage(".*Pravega Sandbox is running.*", 1)
40+
.withStartupTimeout(Duration.ofMinutes(5))
41+
);
42+
43+
private static File driverConfigFile;
44+
45+
@BeforeAll
46+
static void setupDriver() throws Exception {
47+
pravega.start();
48+
var host = pravega.getHost();
49+
50+
log.info("Pravega controller URI: {}", pravega.getControllerURI());
51+
log.info("Pravega segment store: {}:{}", host, pravega.getSegmentStoreEndpoint());
52+
53+
verifyPravegaConnection();
54+
55+
// Fixed config with proper structure from reference file
56+
String driverConfig = """
57+
name: Pravega
58+
driverClass: io.openmessaging.benchmark.driver.pravega.PravegaBenchmarkDriver
59+
60+
client:
61+
controllerURI: %s
62+
scopeName: %s
63+
64+
writer:
65+
enableConnectionPooling: true
66+
67+
includeTimestampInEvent: false
68+
""".formatted(pravega.getControllerURI(), SCOPE);
69+
70+
Path configPath = Files.createTempFile("pravega-driver-", ".yaml");
71+
Files.writeString(configPath, driverConfig);
72+
driverConfigFile = configPath.toFile();
73+
driverConfigFile.deleteOnExit();
74+
75+
log.info("Created Pravega driver config at: {}", driverConfigFile.getAbsolutePath());
76+
log.info("Config content:\n{}", driverConfig);
77+
}
78+
79+
80+
private static void verifyPravegaConnection() {
81+
var clientConfig = ClientConfig.builder()
82+
.controllerURI(URI.create(pravega.getControllerURI()))
83+
.build();
84+
try (StreamManager streamManager = StreamManager.create(clientConfig)) {
85+
boolean created = streamManager.createScope(SCOPE);
86+
log.info("Scope '{}' ensure/create result: {}", SCOPE, created);
87+
}
88+
}
89+
90+
@AfterAll
91+
static void tearDownDriver() {
92+
if (driverConfigFile != null && driverConfigFile.exists()) {
93+
driverConfigFile.delete();
94+
}
95+
}
96+
97+
@Test
98+
void testSimpleProduceConsume() throws Exception {
99+
Workload workload = createSimpleWorkload();
100+
workload.topics = 1;
101+
workload.partitionsPerTopic = 2;
102+
workload.producersPerTopic = 2;
103+
workload.consumerPerSubscription = 2;
104+
workload.producerRate = 500;
105+
workload.useRandomizedPayloads = true;
106+
workload.randomizedPayloadPoolSize = 10;
107+
workload.randomBytesRatio = 0.5;
108+
workload.testDurationMinutes = 1;
109+
110+
TestResult result = runBenchmark(workload);
111+
112+
// Debug logging before validation
113+
log.info("Test result - Topics: {}", result.topics);
114+
log.info("Publish rate size: {}, values: {}", result.publishRate.size(), result.publishRate);
115+
log.info("Consume rate size: {}, values: {}", result.consumeRate.size(), result.consumeRate);
116+
log.info("Aggregate publish: {} msg/s", result.publishRate.stream().reduce(0.0, Double::sum));
117+
118+
validateResults(result, workload);
119+
120+
var total = result.publishRate.stream().reduce(0.0, Double::sum);
121+
assertThat(total)
122+
.as("Should publish expected volume")
123+
.isGreaterThan(workload.producerRate * workload.testDurationMinutes * 0.7);
124+
}
125+
126+
@Test
127+
void testMultipleStreams() throws Exception {
128+
Workload workload = createSimpleWorkload();
129+
workload.topics = 3;
130+
workload.partitionsPerTopic = 2;
131+
workload.producerRate = 300;
132+
workload.useRandomizedPayloads = true;
133+
workload.randomizedPayloadPoolSize = 10;
134+
workload.randomBytesRatio = 0.5;
135+
workload.testDurationMinutes = 1;
136+
137+
TestResult result = runBenchmark(workload);
138+
validateResults(result, workload);
139+
assertThat(result.topics).isEqualTo(workload.topics);
140+
}
141+
142+
@Test
143+
void testHighThroughput() throws Exception {
144+
Workload workload = createSimpleWorkload();
145+
workload.topics = 1;
146+
workload.partitionsPerTopic = 4;
147+
workload.messageSize = 200;
148+
workload.producersPerTopic = 4;
149+
workload.consumerPerSubscription = 4;
150+
workload.producerRate = 800;
151+
workload.useRandomizedPayloads = true;
152+
workload.randomizedPayloadPoolSize = 10;
153+
workload.randomBytesRatio = 0.5;
154+
workload.testDurationMinutes = 1;
155+
156+
TestResult result = runBenchmark(workload);
157+
validateResults(result, workload);
158+
159+
double total = result.publishRate.stream().reduce(0.0, Double::sum);
160+
double actualRate = total / workload.testDurationMinutes;
161+
log.info("Actual Pravega publish rate: {} msg/s", actualRate);
162+
assertThat(actualRate).isGreaterThan(workload.producerRate * 0.5);
163+
}
164+
165+
private TestResult runBenchmark(Workload workload) throws Exception {
166+
var worker = new LocalWorker(statsLogger);
167+
try {
168+
worker.initializeDriver(driverConfigFile);
169+
WorkloadGenerator generator = new WorkloadGenerator("Pravega", workload, worker);
170+
TestResult result = generator.run();
171+
generator.close();
172+
return result;
173+
} finally {
174+
worker.stopAll();
175+
worker.close();
176+
}
177+
}
178+
}
179+
180+

0 commit comments

Comments
 (0)