wyl/hertzbeat
1
0
Fork 0
mirror of https://github.com/dromara/hertzbeat.git synced 2026-09-17 09:40:58 +00:00
Code Issues Projects Releases Wiki Activity

test: launch observability e2e through startup boundary

Browse Source
This commit is contained in:
Logic
2026-08-26 20:45:09 +08:00
parent 34e05a05ce
commit d6a801e36f
15 changed files with 382 additions and 341 deletions
Download Patch File Download Diff File Expand all files Collapse all files
hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/alert/LogPeriodicAlertE2eTest.java
+14 -49
View File
@@ -23,6 +23,9 @@ import org.apache.hertzbeat.alert.reduce.AlarmCommonReduce;
import org.apache.hertzbeat.common.constants.CommonConstants;
import org.apache.hertzbeat.common.entity.alerter.AlertDefine;
import org.apache.hertzbeat.common.entity.alerter.SingleAlert;
import org.apache.hertzbeat.observability.fixture.GreptimeE2eSupport;
import org.apache.hertzbeat.observability.fixture.VectorE2eContainer;
import org.apache.hertzbeat.startup.TrustedStartup;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInstance;
@@ -30,15 +33,9 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.web.server.LocalServerPort;
import org.springframework.test.context.bean.override.mockito.MockitoSpyBean;
import org.springframework.test.context.DynamicPropertyRegistry;
import org.springframework.test.context.DynamicPropertySource;
import org.springframework.test.context.TestPropertySource;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.Network;
import org.testcontainers.containers.wait.strategy.Wait;
import org.testcontainers.Testcontainers;
import org.testcontainers.utility.DockerImageName;
import org.testcontainers.utility.MountableFile;
import java.time.Duration;
import java.util.ArrayList;
@@ -61,21 +58,14 @@ import static org.mockito.Mockito.doAnswer;
* E2E tests for periodic log alert processing.
*/
@SpringBootTest(classes = org.apache.hertzbeat.startup.HertzBeatApplication.class, webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@TrustedStartup
@TestPropertySource(properties = {
"warehouse.store.duckdb.enabled=false",
"warehouse.store.greptime.enabled=true"
"warehouse.store.duckdb.enabled=false"
})
@Slf4j
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
public class LogPeriodicAlertE2eTest {
public class LogPeriodicAlertE2eTest extends GreptimeE2eSupport {
private static final String VECTOR_IMAGE = "timberio/vector:latest-alpine";
private static final int VECTOR_PORT = 8686;
private static final String VECTOR_CONFIG_PATH = "/etc/vector/vector.yml";
private static final String ENV_HERTZBEAT_PORT = "HERTZBEAT_PORT";
private static final String GREPTIME_IMAGE = "greptime/greptimedb:latest";
private static final int GREPTIME_HTTP_PORT = 4000;
private static final int GREPTIME_GRPC_PORT = 4001;
private static final Duration CONTAINER_STARTUP_TIMEOUT = Duration.ofSeconds(120);
@LocalServerPort
@@ -92,30 +82,9 @@ public class LogPeriodicAlertE2eTest {
static GenericContainer<?> vector;
static GenericContainer<?> greptimedb;
static {
greptimedb = new GenericContainer<>(DockerImageName.parse(GREPTIME_IMAGE))
.withExposedPorts(GREPTIME_HTTP_PORT, GREPTIME_GRPC_PORT)
.withCommand("standalone", "start",
"--http-addr", "0.0.0.0:" + GREPTIME_HTTP_PORT,
"--rpc-bind-addr", "0.0.0.0:" + GREPTIME_GRPC_PORT)
.waitingFor(Wait.forListeningPorts(GREPTIME_HTTP_PORT, GREPTIME_GRPC_PORT))
.withStartupTimeout(CONTAINER_STARTUP_TIMEOUT);
greptimedb.start();
}
@DynamicPropertySource
static void greptimeProps(DynamicPropertyRegistry r) {
// Configure GreptimeDB storage endpoints (dynamic ports)
r.add("warehouse.store.greptime.http-endpoint", () -> "http://localhost:" + greptimedb.getMappedPort(GREPTIME_HTTP_PORT));
r.add("warehouse.store.greptime.grpc-endpoints", () -> "localhost:" + greptimedb.getMappedPort(GREPTIME_GRPC_PORT));
r.add("warehouse.store.greptime.username", () -> "");
r.add("warehouse.store.greptime.password", () -> "");
}
@BeforeAll
void setUpAll() throws InterruptedException {
initializeAdministrator();
// Setup test alert definitions
setupTestAlertDefines();
Testcontainers.exposeHostPorts(port);
@@ -124,15 +93,9 @@ public class LogPeriodicAlertE2eTest {
log.info("Waiting for HertzBeat to be fully ready on port {}...", port);
Thread.sleep(5000); // Give HertzBeat time to fully initialize
vector = new GenericContainer<>(DockerImageName.parse(VECTOR_IMAGE))
.withExposedPorts(VECTOR_PORT)
.withCopyFileToContainer(MountableFile.forClasspathResource("vector.yml"), VECTOR_CONFIG_PATH)
.withCommand("--config", "/etc/vector/vector.yml", "--verbose")
.withLogConsumer(outputFrame -> log.info("Vector: {}", outputFrame.getUtf8String()))
.withNetwork(Network.newNetwork())
.withEnv(ENV_HERTZBEAT_PORT, String.valueOf(port))
.waitingFor(Wait.forListeningPort())
.withStartupTimeout(CONTAINER_STARTUP_TIMEOUT);
vector = VectorE2eContainer.create(
port, CONTAINER_STARTUP_TIMEOUT,
outputFrame -> log.info("Vector: {}", outputFrame.getUtf8String()));
vector.start();
}
@@ -207,7 +170,8 @@ public class LogPeriodicAlertE2eTest {
.id(10L)
.name("periodic_error_count_alert_group")
.type(CommonConstants.LOG_ALERT_THRESHOLD_TYPE_PERIODIC)
.expr("SELECT COUNT(*) as error_count FROM hertzbeat_logs WHERE time_unix_nano > NOW() - INTERVAL '10 minute'")
.expr("SELECT COUNT(*) as error_count FROM hertzbeat_logs "
+ "WHERE timestamp > NOW() - INTERVAL '10 minutes'")
.period(10) // Faster schedule for tests
.template("High error count detected: {{ error_count }} errors in last period")
.datasource("sql")
@@ -226,7 +190,8 @@ public class LogPeriodicAlertE2eTest {
.name("periodic_error_count_alert_individual")
.type(CommonConstants.LOG_ALERT_THRESHOLD_TYPE_PERIODIC)
.expr("SELECT COUNT(*) as error_count, severity_text FROM hertzbeat_logs "
+ "WHERE severity_text = 'ERROR' AND time_unix_nano > NOW() - INTERVAL '5 minute' GROUP BY severity_text HAVING COUNT(*) > 2")
+ "WHERE severity_text = 'ERROR' AND timestamp > NOW() - INTERVAL '5 minutes' "
+ "GROUP BY severity_text HAVING COUNT(*) > 2")
.period(10) // Faster schedule for tests
.template("High error count detected: {{ error_count }} errors in last period")
.datasource("sql")
hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/alert/LogRealTimeAlertE2eTest.java
+9 -18
View File
@@ -23,6 +23,9 @@ import org.apache.hertzbeat.common.cache.CacheFactory;
import org.apache.hertzbeat.common.constants.CommonConstants;
import org.apache.hertzbeat.common.entity.alerter.AlertDefine;
import org.apache.hertzbeat.common.entity.alerter.SingleAlert;
import org.apache.hertzbeat.observability.fixture.GreptimeE2eSupport;
import org.apache.hertzbeat.observability.fixture.VectorE2eContainer;
import org.apache.hertzbeat.startup.TrustedStartup;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInstance;
@@ -30,11 +33,7 @@ import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.web.server.LocalServerPort;
import org.springframework.test.context.bean.override.mockito.MockitoSpyBean;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.Network;
import org.testcontainers.containers.wait.strategy.Wait;
import org.testcontainers.Testcontainers;
import org.testcontainers.utility.DockerImageName;
import org.testcontainers.utility.MountableFile;
import java.time.Duration;
import java.util.ArrayList;
@@ -56,14 +55,11 @@ import static org.mockito.Mockito.doAnswer;
* E2E tests for real-time log alert processing.
*/
@SpringBootTest(classes = org.apache.hertzbeat.startup.HertzBeatApplication.class, webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@TrustedStartup
@Slf4j
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
public class LogRealTimeAlertE2eTest {
public class LogRealTimeAlertE2eTest extends GreptimeE2eSupport {
private static final String VECTOR_IMAGE = "timberio/vector:latest-alpine";
private static final int VECTOR_PORT = 8686;
private static final String VECTOR_CONFIG_PATH = "/etc/vector/vector.yml";
private static final String ENV_HERTZBEAT_PORT = "HERTZBEAT_PORT";
private static final Duration CONTAINER_STARTUP_TIMEOUT = Duration.ofSeconds(180);
private static final Duration TEST_WAIT_TIMEOUT = Duration.ofSeconds(120);
@@ -80,6 +76,7 @@ public class LogRealTimeAlertE2eTest {
@BeforeAll
void setUpAll() throws InterruptedException {
initializeAdministrator();
// Setup test alert definitions
setupTestAlertDefines();
@@ -90,15 +87,9 @@ public class LogRealTimeAlertE2eTest {
log.info("Waiting for HertzBeat to be fully ready on port {}...", port);
Thread.sleep(5000); // Give HertzBeat time to fully initialize
vector = new GenericContainer<>(DockerImageName.parse(VECTOR_IMAGE))
.withExposedPorts(VECTOR_PORT)
.withCopyFileToContainer(MountableFile.forClasspathResource("vector.yml"), VECTOR_CONFIG_PATH)
.withCommand("--config", "/etc/vector/vector.yml", "--verbose")
.withLogConsumer(outputFrame -> log.info("Vector: {}", outputFrame.getUtf8String()))
.withNetwork(Network.newNetwork())
.withEnv(ENV_HERTZBEAT_PORT, String.valueOf(port))
.waitingFor(Wait.forListeningPort())
.withStartupTimeout(CONTAINER_STARTUP_TIMEOUT);
vector = VectorE2eContainer.create(
port, CONTAINER_STARTUP_TIMEOUT,
outputFrame -> log.info("Vector: {}", outputFrame.getUtf8String()));
vector.start();
}
hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/fixture/GreptimeE2eSupport.java
+47
View File
@@ -0,0 +1,47 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0.
*/
package org.apache.hertzbeat.observability.fixture;
import java.time.Duration;
import org.apache.hertzbeat.startup.TrustedStartupSupport;
import org.springframework.test.context.DynamicPropertyRegistry;
import org.springframework.test.context.DynamicPropertySource;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.wait.strategy.Wait;
import org.testcontainers.utility.DockerImageName;
/** Shared pinned GreptimeDB data-plane fixture for full-application observability E2E tests. */
public abstract class GreptimeE2eSupport extends TrustedStartupSupport {
private static final int HTTP_PORT = 4000;
private static final int GRPC_PORT = 4001;
@SuppressWarnings("resource")
protected static final GenericContainer<?> GREPTIME = new GenericContainer<>(
DockerImageName.parse("greptime/greptimedb:v1.0.1"))
.withExposedPorts(HTTP_PORT, GRPC_PORT)
.withCommand("standalone", "start",
"--http-addr", "0.0.0.0:" + HTTP_PORT,
"--rpc-bind-addr", "0.0.0.0:" + GRPC_PORT)
.waitingFor(Wait.forListeningPorts(HTTP_PORT, GRPC_PORT))
.withStartupTimeout(Duration.ofSeconds(120));
static {
GREPTIME.start();
}
@DynamicPropertySource
static void greptimeProperties(DynamicPropertyRegistry registry) {
registry.add("warehouse.store.greptime.http-endpoint", () -> "http://" + GREPTIME.getHost()
+ ":" + GREPTIME.getMappedPort(HTTP_PORT));
registry.add("warehouse.store.greptime.grpc-endpoints", () -> GREPTIME.getHost()
+ ":" + GREPTIME.getMappedPort(GRPC_PORT));
registry.add("warehouse.store.greptime.username", () -> "");
registry.add("warehouse.store.greptime.password", () -> "");
}
}
hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/fixture/VectorE2eContainer.java
+55
View File
@@ -0,0 +1,55 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0.
*/
package org.apache.hertzbeat.observability.fixture;
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.time.Duration;
import java.util.function.Consumer;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.output.OutputFrame;
import org.testcontainers.containers.wait.strategy.Wait;
import org.testcontainers.images.builder.Transferable;
import org.testcontainers.utility.DockerImageName;
/** Builds the pinned Vector fixture with a fully rendered, non-templated HertzBeat endpoint. */
public final class VectorE2eContainer {
private static final String IMAGE = "timberio/vector:0.57.0-alpine";
private static final String CONFIG_RESOURCE = "vector.yml";
private static final String CONFIG_PATH = "/etc/vector/vector.yml";
private static final String PORT_PLACEHOLDER = "__HERTZBEAT_PORT__";
private static final int API_PORT = 8686;
private VectorE2eContainer() {
}
public static GenericContainer<?> create(
int hertzbeatPort, Duration startupTimeout, Consumer<OutputFrame> logConsumer) {
String config = readConfig().replace(PORT_PLACEHOLDER, Integer.toString(hertzbeatPort));
return new GenericContainer<>(DockerImageName.parse(IMAGE))
.withExposedPorts(API_PORT)
.withCopyToContainer(Transferable.of(config), CONFIG_PATH)
.withCommand("--config", CONFIG_PATH, "--verbose")
.withLogConsumer(logConsumer)
.waitingFor(Wait.forListeningPort())
.withStartupTimeout(startupTimeout);
}
private static String readConfig() {
try (InputStream input = VectorE2eContainer.class.getClassLoader().getResourceAsStream(CONFIG_RESOURCE)) {
if (input == null) {
throw new IllegalStateException("Vector E2E configuration is missing");
}
return new String(input.readAllBytes(), StandardCharsets.UTF_8);
} catch (IOException failure) {
throw new IllegalStateException("Cannot read Vector E2E configuration", failure);
}
}
}
hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/ingestion/LogIngestionE2eTest.java
+9 -18
View File
@@ -20,6 +20,9 @@ package org.apache.hertzbeat.observability.ingestion;
import lombok.extern.slf4j.Slf4j;
import org.apache.hertzbeat.common.entity.log.LogEntry;
import org.apache.hertzbeat.common.queue.CommonDataQueue;
import org.apache.hertzbeat.observability.fixture.GreptimeE2eSupport;
import org.apache.hertzbeat.observability.fixture.VectorE2eContainer;
import org.apache.hertzbeat.startup.TrustedStartup;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.TestInstance;
@@ -27,11 +30,7 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.web.server.LocalServerPort;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.Network;
import org.testcontainers.containers.wait.strategy.Wait;
import org.testcontainers.Testcontainers;
import org.testcontainers.utility.DockerImageName;
import org.testcontainers.utility.MountableFile;
import java.time.Duration;
import java.util.ArrayList;
@@ -45,14 +44,11 @@ import static org.junit.jupiter.api.Assertions.assertNotNull;
* E2E tests for log ingestion.
*/
@SpringBootTest(classes = org.apache.hertzbeat.startup.HertzBeatApplication.class, webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@TrustedStartup
@Slf4j
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
public class LogIngestionE2eTest {
public class LogIngestionE2eTest extends GreptimeE2eSupport {
private static final String VECTOR_IMAGE = "timberio/vector:latest-alpine";
private static final int VECTOR_PORT = 8686;
private static final String VECTOR_CONFIG_PATH = "/etc/vector/vector.yml";
private static final String ENV_HERTZBEAT_PORT = "HERTZBEAT_PORT";
private static final Duration CONTAINER_STARTUP_TIMEOUT = Duration.ofSeconds(120);
@LocalServerPort
@@ -65,21 +61,16 @@ public class LogIngestionE2eTest {
@BeforeAll
void setUpAll() throws InterruptedException {
initializeAdministrator();
Testcontainers.exposeHostPorts(port);
// Wait for HertzBeat to be fully ready before starting Vector
log.info("Waiting for HertzBeat to be fully ready on port {}...", port);
Thread.sleep(5000); // Give HertzBeat time to fully initialize
vector = new GenericContainer<>(DockerImageName.parse(VECTOR_IMAGE))
.withExposedPorts(VECTOR_PORT)
.withCopyFileToContainer(MountableFile.forClasspathResource("vector.yml"), VECTOR_CONFIG_PATH)
.withCommand("--config", "/etc/vector/vector.yml", "--verbose")
.withLogConsumer(outputFrame -> log.info("Vector: {}", outputFrame.getUtf8String()))
.withNetwork(Network.newNetwork())
.withEnv(ENV_HERTZBEAT_PORT, String.valueOf(port))
.waitingFor(Wait.forListeningPort())
.withStartupTimeout(CONTAINER_STARTUP_TIMEOUT);
vector = VectorE2eContainer.create(
port, CONTAINER_STARTUP_TIMEOUT,
outputFrame -> log.info("Vector: {}", outputFrame.getUtf8String()));
vector.start();
}
hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/storage/AuthenticatedGreptimeThreeSignalPublicApiE2eTest.java
+4 -33
View File
@@ -33,15 +33,7 @@ import java.time.Duration;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import org.apache.hertzbeat.common.constants.CommonConstants;
import org.apache.hertzbeat.common.entity.manager.Collector;
import org.apache.hertzbeat.manager.dao.CollectorDao;
import org.apache.hertzbeat.manager.instrumentation.intake.CollectorIntakeAdvertisementCodec;
import org.apache.hertzbeat.manager.instrumentation.intake.CollectorIntakeAdvertisementRequest;
import org.apache.hertzbeat.manager.pojo.dto.CollectorInstrumentationIntake.Capability;
import org.apache.hertzbeat.manager.pojo.dto.CollectorInstrumentationIntake.Gateway;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.web.server.LocalServerPort;
import org.testcontainers.junit.jupiter.Testcontainers;
@@ -58,7 +50,6 @@ import org.testcontainers.junit.jupiter.Testcontainers;
"scheduler.server.enabled=false",
"spring.datasource.url=jdbc:h2:mem:hertzbeat-authenticated-greptime-e2e;MODE=MYSQL;DB_CLOSE_DELAY=-1",
"warehouse.store.duckdb.enabled=false",
"warehouse.store.greptime.enabled=true",
"warehouse.store.greptime.username=",
"warehouse.store.greptime.password="
})
@@ -75,11 +66,9 @@ class AuthenticatedGreptimeThreeSignalPublicApiE2eTest extends GreptimeThreeSign
@LocalServerPort
private int serverPort;
@Autowired
private CollectorDao collectorDao;
@Test
void authenticatedPublicApiIngestsDetectsAndQueriesThreeSignalsInGreptime() throws Exception {
initializeAdministrator();
advertiseCollectorProfile();
long startedAt = System.currentTimeMillis() - 1_000;
long signalTimeNanos = System.currentTimeMillis() * 1_000_000L;
@@ -180,7 +169,9 @@ class AuthenticatedGreptimeThreeSignalPublicApiE2eTest extends GreptimeThreeSign
HttpResponse<byte[]> response = send(postJson("/api/instrumentation/detect", detectionBody, adminToken));
assertThat(response.statusCode()).isEqualTo(200);
JsonNode envelope = OBJECT_MAPPER.readTree(response.body());
assertThat(envelope.path("code").asInt()).isZero();
assertThat(envelope.path("code").asInt())
.as("scoped detection response: %s", envelope)
.isZero();
JsonNode data = envelope.path("data");
assertThat(data.path("signals").path("metrics").path("status").asText()).isNotEqualTo("received");
assertThat(data.path("signals").path("logs").path("status").asText()).isNotEqualTo("received");
@@ -407,26 +398,6 @@ class AuthenticatedGreptimeThreeSignalPublicApiE2eTest extends GreptimeThreeSign
"primaryIdentity", primary);
}
/**
* Collector persistence is deterministic test setup only. All behavior under proof starts at the public HTTP
* boundary; no ingestion, detection, or query service is invoked directly.
*/
private void advertiseCollectorProfile() {
String advertisement = new CollectorIntakeAdvertisementCodec().encode(
new CollectorIntakeAdvertisementRequest(
1,
Gateway.COLLECTOR,
List.of(Capability.OTLP_HTTP_PROTOBUF),
"http://127.0.0.1:4318",
null));
collectorDao.save(Collector.builder()
.name(COLLECTOR_ID)
.ip("127.0.0.1")
.status(CommonConstants.COLLECTOR_STATUS_ONLINE)
.instrumentationIntake(advertisement)
.build());
}
private HttpRequest postProtobuf(String path, byte[] body, String token) {
HttpRequest.Builder builder = request(path)
.header("Content-Type", "application/x-protobuf")
hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/storage/GreptimeLogStorageE2eTest.java
+26 -99
View File
@@ -19,7 +19,9 @@ package org.apache.hertzbeat.observability.storage;
import lombok.extern.slf4j.Slf4j;
import org.apache.hertzbeat.common.entity.log.LogEntry;
import org.apache.hertzbeat.common.queue.CommonDataQueue;
import org.apache.hertzbeat.observability.fixture.GreptimeE2eSupport;
import org.apache.hertzbeat.observability.fixture.VectorE2eContainer;
import org.apache.hertzbeat.startup.TrustedStartup;
import org.apache.hertzbeat.warehouse.store.history.tsdb.greptime.GreptimeDbDataStorage;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
@@ -27,20 +29,14 @@ import org.junit.jupiter.api.TestInstance;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.web.server.LocalServerPort;
import org.springframework.test.context.DynamicPropertyRegistry;
import org.springframework.test.context.DynamicPropertySource;
import org.springframework.test.context.TestPropertySource;
import java.util.ArrayList;
import java.util.List;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.Network;
import org.testcontainers.containers.wait.strategy.Wait;
import org.testcontainers.Testcontainers;
import org.testcontainers.utility.DockerImageName;
import org.testcontainers.utility.MountableFile;
import java.time.Duration;
import java.util.Collections;
import java.util.List;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.Testcontainers;
import static org.awaitility.Awaitility.await;
import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -50,57 +46,29 @@ import static org.junit.jupiter.api.Assertions.assertNotNull;
* E2E tests for GreptimeDB log storage.
*/
@SpringBootTest(classes = org.apache.hertzbeat.startup.HertzBeatApplication.class, webEnvironment = SpringBootTest.WebEnvironment.RANDOM_PORT)
@TrustedStartup
@TestPropertySource(properties = {
"warehouse.store.duckdb.enabled=false",
"warehouse.store.greptime.enabled=true"
"warehouse.store.duckdb.enabled=false"
})
@Slf4j
@TestInstance(TestInstance.Lifecycle.PER_CLASS)
public class GreptimeLogStorageE2eTest {
public class GreptimeLogStorageE2eTest extends GreptimeE2eSupport {
private static final String VECTOR_IMAGE = "timberio/vector:latest-alpine";
private static final int VECTOR_PORT = 8686;
private static final String VECTOR_CONFIG_PATH = "/etc/vector/vector.yml";
private static final String ENV_HERTZBEAT_PORT = "HERTZBEAT_PORT";
private static final String GREPTIME_IMAGE = "greptime/greptimedb:latest";
private static final int GREPTIME_HTTP_PORT = 4000;
private static final int GREPTIME_GRPC_PORT = 4001;
private static final Duration CONTAINER_STARTUP_TIMEOUT = Duration.ofSeconds(120);
private static final String VECTOR_SERVICE_NAME = "hertzbeat-vector-e2e";
@LocalServerPort
private int port;
@Autowired
private CommonDataQueue commonDataQueue;
@Autowired
private GreptimeDbDataStorage greptimeDbDataStorage;
static GenericContainer<?> vector;
static GenericContainer<?> greptimedb;
static {
greptimedb = new GenericContainer<>(DockerImageName.parse(GREPTIME_IMAGE))
.withExposedPorts(GREPTIME_HTTP_PORT, GREPTIME_GRPC_PORT)
.withCommand("standalone", "start",
"--http-addr", "0.0.0.0:" + GREPTIME_HTTP_PORT,
"--rpc-bind-addr", "0.0.0.0:" + GREPTIME_GRPC_PORT)
.waitingFor(Wait.forListeningPorts(GREPTIME_HTTP_PORT, GREPTIME_GRPC_PORT))
.withStartupTimeout(CONTAINER_STARTUP_TIMEOUT);
greptimedb.start();
}
@DynamicPropertySource
static void greptimeProps(DynamicPropertyRegistry r) {
r.add("warehouse.store.greptime.http-endpoint", () -> "http://localhost:" + greptimedb.getMappedPort(GREPTIME_HTTP_PORT));
r.add("warehouse.store.greptime.grpc-endpoints", () -> "localhost:" + greptimedb.getMappedPort(GREPTIME_GRPC_PORT));
r.add("warehouse.store.greptime.username", () -> "");
r.add("warehouse.store.greptime.password", () -> "");
}
private long vectorStartedAtMillis;
@BeforeAll
void setUpAll() throws InterruptedException {
initializeAdministrator();
// Expose host ports for testcontainers
Testcontainers.exposeHostPorts(port);
@@ -108,15 +76,10 @@ public class GreptimeLogStorageE2eTest {
log.info("Waiting for HertzBeat to be fully ready on port {}...", port);
Thread.sleep(5000); // Give HertzBeat time to fully initialize
vector = new GenericContainer<>(DockerImageName.parse(VECTOR_IMAGE))
.withExposedPorts(VECTOR_PORT)
.withCopyFileToContainer(MountableFile.forClasspathResource("vector.yml"), VECTOR_CONFIG_PATH)
.withCommand("--config", "/etc/vector/vector.yml", "--verbose")
.withLogConsumer(outputFrame -> log.info("Vector: {}", outputFrame.getUtf8String()))
.withNetwork(Network.newNetwork())
.withEnv(ENV_HERTZBEAT_PORT, String.valueOf(port))
.waitingFor(Wait.forListeningPort())
.withStartupTimeout(CONTAINER_STARTUP_TIMEOUT);
vectorStartedAtMillis = System.currentTimeMillis();
vector = VectorE2eContainer.create(
port, CONTAINER_STARTUP_TIMEOUT,
outputFrame -> log.info("Vector: {}", outputFrame.getUtf8String()));
vector.start();
}
@@ -124,53 +87,16 @@ public class GreptimeLogStorageE2eTest {
void testLogStorageToGreptimeDb() {
log.info("GreptimeDbDataStorage serverAvailable: {}", greptimeDbDataStorage.isServerAvailable());
List<LogEntry> capturedLogs = new ArrayList<>();
// Wait for Vector to generate and send logs to HertzBeat
await().atMost(Duration.ofSeconds(30))
.pollInterval(Duration.ofSeconds(3))
.untilAsserted(() -> {
// Poll log entries from the queue (non-blocking)
try {
LogEntry logEntry = commonDataQueue.pollLogEntry();
if (logEntry != null) {
capturedLogs.add(logEntry);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
throw new RuntimeException("Test interrupted", e);
}
// Assert that we have captured at least some logs
assertFalse(capturedLogs.isEmpty(), "Should have captured at least one log entry");
});
// Verify the captured logs
assertFalse(capturedLogs.isEmpty(), "No logs were captured from Vector");
LogEntry firstLog = capturedLogs.get(0);
assertNotNull(firstLog, "First log should not be null");
assertNotNull(firstLog.getBody(), "Log body should not be null");
assertNotNull(firstLog.getSeverityText(), "Severity text should not be null");
// Directly write logs to GreptimeDB to test storage functionality
log.info("Directly writing {} captured logs to GreptimeDB", capturedLogs.size());
greptimeDbDataStorage.saveLogDataBatch(capturedLogs);
// Give some time for the write to complete
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
// Additional wait to ensure logs are persisted to GreptimeDB
await().atMost(Duration.ofSeconds(30))
// The production OTLP path writes directly to GreptimeDB; verify the persisted rows.
await().atMost(Duration.ofSeconds(60))
.pollInterval(Duration.ofSeconds(2))
.untilAsserted(() -> {
// Query GreptimeDB directly to verify data persistence
List<LogEntry> storedLogs = queryStoredLogs();
log.info("Queried {} logs from GreptimeDB", storedLogs.size());
assertFalse(storedLogs.isEmpty(), "Should have logs stored in GreptimeDB");
LogEntry firstLog = storedLogs.get(0);
assertNotNull(firstLog.getBody(), "Stored log body should not be null");
assertNotNull(firstLog.getSeverityText(), "Stored log severity should not be null");
});
}
@@ -178,9 +104,10 @@ public class GreptimeLogStorageE2eTest {
* Helper method to query stored logs directly from GreptimeDB
*/
private List<LogEntry> queryStoredLogs() {
// Query without time condition to verify data exists
// Scope the query to this Vector run so rows from other E2E tests cannot satisfy the proof.
List<LogEntry> result = greptimeDbDataStorage.queryLogsByMultipleConditions(
null, null, null, null, null, null, null);
vectorStartedAtMillis, System.currentTimeMillis(), null, null, null, null, null,
Collections.emptySet(), true, null, VECTOR_SERVICE_NAME, null, null);
log.info("queryLogsByMultipleConditions returned {} entries", result.size());
return result;
}
hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/storage/GreptimeThreeSignalE2eSupport.java
+53 -26
View File
@@ -35,18 +35,25 @@ import io.opentelemetry.proto.resource.v1.Resource;
import io.opentelemetry.proto.trace.v1.ResourceSpans;
import io.opentelemetry.proto.trace.v1.ScopeSpans;
import io.opentelemetry.proto.trace.v1.Span;
import java.time.Duration;
import java.time.Instant;
import java.util.HexFormat;
import java.util.List;
import org.springframework.test.context.DynamicPropertyRegistry;
import org.springframework.test.context.DynamicPropertySource;
import org.testcontainers.containers.GenericContainer;
import org.testcontainers.containers.wait.strategy.Wait;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.utility.DockerImageName;
import org.apache.hertzbeat.common.constants.CommonConstants;
import org.apache.hertzbeat.common.entity.dto.ManagedOtelRuntimeStatus;
import org.apache.hertzbeat.common.entity.manager.Collector;
import org.apache.hertzbeat.manager.dao.CollectorDao;
import org.apache.hertzbeat.manager.instrumentation.intake.CollectorIntakeAdvertisementCodec;
import org.apache.hertzbeat.manager.instrumentation.intake.CollectorIntakeAdvertisementRequest;
import org.apache.hertzbeat.manager.pojo.dto.CollectorInstrumentationIntake.Capability;
import org.apache.hertzbeat.manager.pojo.dto.CollectorInstrumentationIntake.Gateway;
import org.apache.hertzbeat.manager.scheduler.runtime.CollectorRuntimeStatusRegistry;
import org.apache.hertzbeat.observability.fixture.GreptimeE2eSupport;
import org.apache.hertzbeat.startup.TrustedStartup;
import org.springframework.beans.factory.annotation.Autowired;
/** Shared real-Greptime container and deterministic OTLP payloads for three-signal E2E tests. */
abstract class GreptimeThreeSignalE2eSupport {
@TrustedStartup
abstract class GreptimeThreeSignalE2eSupport extends GreptimeE2eSupport {
static final String SERVICE_NAME = "checkout-api";
static final String SERVICE_NAMESPACE = "commerce";
@@ -62,26 +69,46 @@ abstract class GreptimeThreeSignalE2eSupport {
static final String LOG_BODY = "three-signal-e2e";
static final String SPAN_NAME = "GET /checkout";
private static final int GREPTIME_HTTP_PORT = 4000;
private static final int GREPTIME_GRPC_PORT = 4001;
@Autowired
private CollectorDao collectorDao;
@Container
@SuppressWarnings("resource")
static final GenericContainer<?> GREPTIME = new GenericContainer<>(
DockerImageName.parse("greptime/greptimedb:v1.0.1"))
.withExposedPorts(GREPTIME_HTTP_PORT, GREPTIME_GRPC_PORT)
.withCommand("standalone", "start",
"--http-addr", "0.0.0.0:" + GREPTIME_HTTP_PORT,
"--rpc-bind-addr", "0.0.0.0:" + GREPTIME_GRPC_PORT)
.waitingFor(Wait.forListeningPorts(GREPTIME_HTTP_PORT, GREPTIME_GRPC_PORT))
.withStartupTimeout(Duration.ofSeconds(120));
@Autowired
private CollectorRuntimeStatusRegistry collectorRuntimeStatusRegistry;
@DynamicPropertySource
static void greptimeProperties(DynamicPropertyRegistry registry) {
registry.add("warehouse.store.greptime.http-endpoint", () -> "http://" + GREPTIME.getHost()
+ ":" + GREPTIME.getMappedPort(GREPTIME_HTTP_PORT));
registry.add("warehouse.store.greptime.grpc-endpoints", () -> GREPTIME.getHost()
+ ":" + GREPTIME.getMappedPort(GREPTIME_GRPC_PORT));
final void advertiseCollectorProfile() {
String advertisement = new CollectorIntakeAdvertisementCodec().encode(
new CollectorIntakeAdvertisementRequest(
1,
Gateway.COLLECTOR,
List.of(Capability.OTLP_HTTP_PROTOBUF),
"http://127.0.0.1:4318",
null));
collectorDao.save(Collector.builder()
.name(COLLECTOR_ID)
.ip("127.0.0.1")
.status(CommonConstants.COLLECTOR_STATUS_ONLINE)
.instrumentationIntake(advertisement)
.build());
collectorRuntimeStatusRegistry.report(COLLECTOR_ID, availableCollectorGateway());
}
private ManagedOtelRuntimeStatus availableCollectorGateway() {
return new ManagedOtelRuntimeStatus(
ManagedOtelRuntimeStatus.CURRENT_SCHEMA_VERSION,
true,
ManagedOtelRuntimeStatus.RuntimeState.RUNNING,
1,
1,
-1,
ManagedOtelRuntimeStatus.IntakeCredentialState.CONFIGURED,
0,
Instant.now(),
"",
ManagedOtelRuntimeStatus.FailureCode.NONE,
ManagedOtelRuntimeStatus.RuntimeTelemetry.unavailable(false),
List.of(),
ManagedOtelRuntimeStatus.OtlpGatewayStatus.available(
List.of(ManagedOtelRuntimeStatus.OtlpGatewayTransport.HTTP_PROTOBUF)));
}
static ExportMetricsServiceRequest metrics(long timeNanos) {
hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/storage/GreptimeThreeSignalInstrumentationE2eTest.java
+30 -52
View File
@@ -27,16 +27,10 @@ import static org.awaitility.Awaitility.await;
import java.time.Duration;
import java.util.List;
import java.util.Map;
import org.apache.hertzbeat.common.constants.CommonConstants;
import org.apache.hertzbeat.common.entity.log.LogEntry;
import org.apache.hertzbeat.common.entity.manager.Collector;
import org.apache.hertzbeat.common.observability.gateway.AuthTokenRequestContext;
import org.apache.hertzbeat.common.observability.dto.metrics.OtlpMetricsConsoleDto;
import org.apache.hertzbeat.common.observability.dto.trace.TraceListItemDto;
import org.apache.hertzbeat.manager.dao.CollectorDao;
import org.apache.hertzbeat.manager.instrumentation.intake.CollectorIntakeAdvertisementCodec;
import org.apache.hertzbeat.manager.instrumentation.intake.CollectorIntakeAdvertisementRequest;
import org.apache.hertzbeat.manager.pojo.dto.CollectorInstrumentationIntake.Capability;
import org.apache.hertzbeat.manager.pojo.dto.CollectorInstrumentationIntake.Gateway;
import org.apache.hertzbeat.observability.ingestion.service.OtlpGrpcIngestionService;
import org.apache.hertzbeat.observability.instrumentation.api.InstrumentationApiContract.DetectionRequest;
import org.apache.hertzbeat.observability.instrumentation.api.InstrumentationApiContract.DetectionResponse;
@@ -74,7 +68,6 @@ import org.testcontainers.junit.jupiter.Testcontainers;
"scheduler.server.enabled=false",
"spring.datasource.url=jdbc:h2:mem:hertzbeat-e2e;MODE=MYSQL;DB_CLOSE_DELAY=-1",
"warehouse.store.duckdb.enabled=false",
"warehouse.store.greptime.enabled=true",
"warehouse.store.greptime.username=",
"warehouse.store.greptime.password="
})
@@ -90,9 +83,6 @@ class GreptimeThreeSignalInstrumentationE2eTest extends GreptimeThreeSignalE2eSu
@Autowired
private InstrumentationDetectionV2Service currentDetectionService;
@Autowired
private CollectorDao collectorDao;
@Autowired
private InstrumentationSignalDetectionStore signalDetectionStore;
@@ -114,9 +104,15 @@ class GreptimeThreeSignalInstrumentationE2eTest extends GreptimeThreeSignalE2eSu
long startedAt = System.currentTimeMillis() - 1_000;
long signalTimeNanos = System.currentTimeMillis() * 1_000_000L;
ingestionService.ingestMetricsGrpc(metrics(signalTimeNanos));
ingestionService.ingestLogsGrpc(logs(signalTimeNanos));
ingestionService.ingestTracesGrpc(traces(signalTimeNanos));
AuthTokenRequestContext.bindWorkspaceId("default");
AuthTokenRequestContext.bindCollectorId(COLLECTOR_ID);
try {
ingestionService.ingestMetricsGrpc(metrics(signalTimeNanos));
ingestionService.ingestLogsGrpc(logs(signalTimeNanos));
ingestionService.ingestTracesGrpc(traces(signalTimeNanos));
} finally {
AuthTokenRequestContext.clear();
}
await().atMost(Duration.ofSeconds(20)).pollInterval(Duration.ofSeconds(1)).untilAsserted(() -> {
List<Map<String, Object>> rows = queryExecutor.executeStrict(
@@ -194,39 +190,15 @@ class GreptimeThreeSignalInstrumentationE2eTest extends GreptimeThreeSignalE2eSu
assertNotReceived(requestWithContext(request, INSTANCE_ID, "/other"));
DetectionResponse detected = detectionService.detect(request);
assertProductionQueries(
enabledJump(detected, METRICS).context(),
enabledJump(detected, LOGS).context(),
enabledJump(detected, TRACES).context());
}
private void advertiseCollectorProfile() {
CollectorIntakeAdvertisementCodec codec = new CollectorIntakeAdvertisementCodec();
String advertisement = codec.encode(
new CollectorIntakeAdvertisementRequest(
1,
Gateway.COLLECTOR,
List.of(Capability.OTLP_HTTP_PROTOBUF),
"http://127.0.0.1:4318",
null));
collectorDao.save(Collector.builder()
.name(COLLECTOR_ID)
.ip("127.0.0.1")
.status(CommonConstants.COLLECTOR_STATUS_ONLINE)
.instrumentationIntake(advertisement)
.build());
String serverAdvertisement = codec.encode(new CollectorIntakeAdvertisementRequest(
1,
Gateway.SERVER,
List.of(Capability.OTLP_HTTP_PROTOBUF),
"http://127.0.0.1:4318",
null));
collectorDao.save(Collector.builder()
.name(SERVER_PROFILE_ID)
.ip("127.0.0.1")
.status(CommonConstants.COLLECTOR_STATUS_ONLINE)
.instrumentationIntake(serverAdvertisement)
.build());
AuthTokenRequestContext.bindWorkspaceId("default");
try {
assertProductionQueries(
enabledJump(detected, METRICS).context(),
enabledJump(detected, LOGS).context(),
enabledJump(detected, TRACES).context());
} finally {
AuthTokenRequestContext.clear();
}
}
private void assertCurrentDetectionContract(long startedAt) {
@@ -265,7 +237,7 @@ class GreptimeThreeSignalInstrumentationE2eTest extends GreptimeThreeSignalE2eSu
Platform.LINUX_AMD64,
new ServiceIdentity(
SERVICE_NAME, SERVICE_NAMESPACE, ENVIRONMENT, INSTANCE_ID, ENDPOINT),
"server:" + SERVER_PROFILE_ID,
SERVER_PROFILE_ID,
startedAt));
assertThat(directServerResponse.signals().values())
.allMatch(signal -> signal.status()
@@ -277,8 +249,11 @@ class GreptimeThreeSignalInstrumentationE2eTest extends GreptimeThreeSignalE2eSu
assertThat(queryExecutor.executeStrict("""
SELECT service_name, resource_attributes, log_attributes, timestamp
FROM hertzbeat_logs
WHERE service_name = '%s'
AND timestamp >= to_timestamp_millis(%d)
AND timestamp < to_timestamp_millis(%d)
ORDER BY timestamp DESC LIMIT 1
"""))
""".formatted(SERVICE_NAME, startedAt, detectedAt + 1)))
.singleElement()
.satisfies(row -> {
assertThat(String.valueOf(row.get("resource_attributes")))
@@ -313,8 +288,11 @@ class GreptimeThreeSignalInstrumentationE2eTest extends GreptimeThreeSignalE2eSu
"span_attributes.http.route" AS http_route,
timestamp
FROM hzb_traces
WHERE service_name = '%s'
AND timestamp >= to_timestamp_millis(%d)
AND timestamp < to_timestamp_millis(%d)
ORDER BY timestamp DESC LIMIT 1
"""))
""".formatted(SERVICE_NAME, startedAt, detectedAt + 1)))
.singleElement()
.satisfies(row -> {
assertThat(row.get("service_instance_id")).hasToString(INSTANCE_ID);
@@ -335,7 +313,7 @@ class GreptimeThreeSignalInstrumentationE2eTest extends GreptimeThreeSignalE2eSu
QueryJumpContext tracesContext) {
long end = metricsContext.detectedAt() + 60_000;
OtlpMetricsConsoleDto metrics = metricsQueryService.query(new CollectorScopedMetricsQueryService.Request(
null, null, null, metricsContext.startedAt(), end, metricsContext.serviceName(),
"default", null, null, metricsContext.startedAt(), end, metricsContext.serviceName(),
metricsContext.serviceNamespace(), metricsContext.environment(), metricsContext.collectorId(),
INSTANCE_ID, ENDPOINT, "hertzbeat_e2e_requests", null, null,
null, null, "1s", "20", null));
@@ -354,7 +332,7 @@ class GreptimeThreeSignalInstrumentationE2eTest extends GreptimeThreeSignalE2eSu
.anySatisfy(row -> assertThat(Double.parseDouble(String.valueOf(row[1]))).isEqualTo(1.0));
OtlpMetricsConsoleDto missingInstanceMetrics = metricsQueryService.query(
new CollectorScopedMetricsQueryService.Request(
null, null, null, metricsContext.startedAt(), end, metricsContext.serviceName(),
"default", null, null, metricsContext.startedAt(), end, metricsContext.serviceName(),
metricsContext.serviceNamespace(), metricsContext.environment(), metricsContext.collectorId(),
"other-instance", ENDPOINT, "hertzbeat_e2e_requests", null, null,
null, null, "1s", "20", null));
hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/storage/PrometheusActiveSourcePublicApiE2eTest.java
+1 -29
View File
@@ -22,16 +22,8 @@ import java.nio.file.Path;
import java.time.Duration;
import java.util.LinkedHashMap;
import java.util.Map;
import org.apache.hertzbeat.common.constants.CommonConstants;
import org.apache.hertzbeat.common.entity.manager.Collector;
import org.apache.hertzbeat.manager.dao.CollectorDao;
import org.apache.hertzbeat.manager.instrumentation.intake.CollectorIntakeAdvertisementCodec;
import org.apache.hertzbeat.manager.instrumentation.intake.CollectorIntakeAdvertisementRequest;
import org.apache.hertzbeat.manager.pojo.dto.CollectorInstrumentationIntake.Capability;
import org.apache.hertzbeat.manager.pojo.dto.CollectorInstrumentationIntake.Gateway;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.io.TempDir;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.boot.test.web.server.LocalServerPort;
import org.testcontainers.junit.jupiter.Testcontainers;
@@ -46,7 +38,6 @@ import org.testcontainers.junit.jupiter.Testcontainers;
"scheduler.server.enabled=false",
"spring.datasource.url=jdbc:h2:mem:hertzbeat-active-prometheus-e2e;MODE=MYSQL;DB_CLOSE_DELAY=-1",
"warehouse.store.duckdb.enabled=false",
"warehouse.store.greptime.enabled=true",
"warehouse.store.greptime.username=",
"warehouse.store.greptime.password="
})
@@ -61,14 +52,12 @@ class PrometheusActiveSourcePublicApiE2eTest extends GreptimeThreeSignalE2eSuppo
@LocalServerPort
private int serverPort;
@Autowired
private CollectorDao collectorDao;
@TempDir
private Path tempDir;
@Test
void managedPrometheusSourceScrapesWritesAndSurfacesPersistedEvidence() throws Exception {
initializeAdministrator();
advertiseCollectorProfile();
String adminToken = login();
long entityId = createEntity(adminToken);
@@ -191,23 +180,6 @@ class PrometheusActiveSourcePublicApiE2eTest extends GreptimeThreeSignalE2eSuppo
return OBJECT_MAPPER.writeValueAsBytes(request);
}
/** Collector persistence is control-plane setup; telemetry starts only at the real scrape endpoint. */
private void advertiseCollectorProfile() {
String advertisement = new CollectorIntakeAdvertisementCodec().encode(
new CollectorIntakeAdvertisementRequest(
1,
Gateway.COLLECTOR,
java.util.List.of(Capability.OTLP_HTTP_PROTOBUF),
"http://127.0.0.1:4318",
null));
collectorDao.save(Collector.builder()
.name(COLLECTOR_ID)
.ip("127.0.0.1")
.status(CommonConstants.COLLECTOR_STATUS_ONLINE)
.instrumentationIntake(advertisement)
.build());
}
private JsonNode successfulJson(HttpResponse<byte[]> response) throws Exception {
assertThat(response.statusCode()).isEqualTo(200);
JsonNode envelope = OBJECT_MAPPER.readTree(response.body());
hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/startup/TrustedStartup.java
+26
View File
@@ -0,0 +1,26 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0.
*/
package org.apache.hertzbeat.startup;
import java.lang.annotation.ElementType;
import java.lang.annotation.Inherited;
import java.lang.annotation.Retention;
import java.lang.annotation.RetentionPolicy;
import java.lang.annotation.Target;
import org.apache.hertzbeat.startup.runtime.TrustedSpringBootContextLoader;
import org.springframework.test.context.ActiveProfiles;
import org.springframework.test.context.ContextConfiguration;
/** Runs an E2E full-application test through the official trusted startup boundary. */
@Target(ElementType.TYPE)
@Retention(RetentionPolicy.RUNTIME)
@Inherited
@ActiveProfiles({"test", "e2e"})
@ContextConfiguration(loader = TrustedSpringBootContextLoader.class)
public @interface TrustedStartup {
}
hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/startup/TrustedStartupSupport.java
+31
View File
@@ -0,0 +1,31 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0.
*/
package org.apache.hertzbeat.startup;
import org.apache.hertzbeat.manager.setup.identity.AdministratorCredentials;
import org.apache.hertzbeat.manager.setup.identity.DatabaseAccountRepository;
import org.apache.hertzbeat.manager.setup.identity.IdentityInitializationService;
import org.springframework.beans.factory.annotation.Autowired;
/** Shared identity fixture for E2E tests that exercise authenticated public boundaries. */
@TrustedStartup
public abstract class TrustedStartupSupport {
@Autowired
private IdentityInitializationService identityInitializationService;
@Autowired
private DatabaseAccountRepository databaseAccountRepository;
protected final void initializeAdministrator() {
if (!databaseAccountRepository.existsByUsername("admin")) {
identityInitializationService.createFirstAdministrator(
new AdministratorCredentials("admin", "hertzbeat".toCharArray()));
}
}
}
hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/startup/runtime/TrustedSpringBootContextLoader.java
+41
View File
@@ -0,0 +1,41 @@
/*
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0.
*/
package org.apache.hertzbeat.startup.runtime;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import org.apache.hertzbeat.manager.setup.runtime.SetupRuntimeTransition;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.test.context.SpringBootContextLoader;
import org.springframework.core.env.StandardEnvironment;
/** Launches E2E Spring tests through the same trusted admission used in production. */
public final class TrustedSpringBootContextLoader extends SpringBootContextLoader {
@Override
protected SpringApplication getSpringApplication() {
Path installationRoot = testInstallationRoot();
StandardEnvironment environment = new StandardEnvironment();
environment.getPropertySources().addFirst(StartupLaunchAdmission.internalPropertySource(
StartupDecision.normal(), installationRoot, StartupLaunchAdmission.Mode.ORDINARY));
SpringApplication application = super.getSpringApplication();
application.setEnvironment(environment);
application.addInitializers(context -> context.getBeanFactory().registerSingleton(
"setupRuntimeTransition", (SetupRuntimeTransition) () -> { }));
return application;
}
private static Path testInstallationRoot() {
try {
return Files.createTempDirectory("hertzbeat-e2e-");
} catch (IOException failure) {
throw new IllegalStateException("Cannot prepare the E2E installation root", failure);
}
}
}
hertzbeat-e2e/hertzbeat-observability-e2e/src/test/resources/application-e2e.yml
+25
View File
@@ -0,0 +1,25 @@
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0.
# Keep managed values classpath-backed so the production setup projection can
# retain honest source attribution when a full application context starts.
spring:
jpa:
hibernate:
ddl-auto: none
flyway:
enabled: true
locations: classpath:db/migration/h2
warehouse:
store:
greptime:
enabled: true
hertzbeat:
instrumentation:
server:
profile-id: server-e2e
otlp-http-endpoint: http://127.0.0.1:4318
authentication: bearer_token
hertzbeat-e2e/hertzbeat-observability-e2e/src/test/resources/vector.yml
+11 -17
View File
@@ -61,7 +61,7 @@ transforms:
"resource": {
"attributes": [
{ "key": "source_type", "value": { "stringValue": .source_type } },
{ "key": "service.name", "value": { "stringValue": syslog.appname } },
{ "key": "service.name", "value": { "stringValue": "hertzbeat-vector-e2e" } },
{ "key": "host.hostname", "value": { "stringValue": syslog.hostname } }
]
},
@@ -102,25 +102,19 @@ sinks:
type: opentelemetry
protocol:
type: http
uri: "http://host.testcontainers.internal:${HERTZBEAT_PORT:-1157}/api/logs/ingest/otlp"
uri: "http://host.testcontainers.internal:__HERTZBEAT_PORT__/api/logs/otlp/v1/logs"
method: post
encoding:
codec: json
framing:
method: newline_delimited
headers:
content-type: application/json
codec: otlp
auth:
strategy: basic
user: admin
password: hertzbeat
# Increase timeout and retry settings for stability in CI environments
request:
timeout_secs: 60
retry_attempts: 10
retry_initial_backoff_secs: 2
retry_max_duration_secs: 120
# Batch settings for better throughput
batch:
max_bytes: 524288
timeout_secs: 5
request:
timeout_secs: 60
retry_attempts: 10
retry_initial_backoff_secs: 2
retry_max_duration_secs: 120
batch:
max_bytes: 524288
timeout_secs: 5