This is an automated email from the ASF dual-hosted git repository. zqr10159 pushed a commit to branch 2.0.0 in repository https://gitbox.apache.org/repos/asf/hertzbeat.git
commit d6a801e36fefb3d0ba3e90289d90bfedc7f0cb69 Author: Logic <[email protected]> AuthorDate: Wed Aug 26 20:45:09 2026 +0800 test: launch observability e2e through startup boundary --- .../alert/LogPeriodicAlertE2eTest.java | 65 +++-------- .../alert/LogRealTimeAlertE2eTest.java | 27 ++--- .../observability/fixture/GreptimeE2eSupport.java | 47 ++++++++ .../observability/fixture/VectorE2eContainer.java | 55 +++++++++ .../ingestion/LogIngestionE2eTest.java | 27 ++--- ...ticatedGreptimeThreeSignalPublicApiE2eTest.java | 37 +------ .../storage/GreptimeLogStorageE2eTest.java | 123 +++++---------------- .../storage/GreptimeThreeSignalE2eSupport.java | 79 ++++++++----- .../GreptimeThreeSignalInstrumentationE2eTest.java | 82 +++++--------- .../PrometheusActiveSourcePublicApiE2eTest.java | 30 +---- .../apache/hertzbeat/startup/TrustedStartup.java | 26 +++++ .../hertzbeat/startup/TrustedStartupSupport.java | 31 ++++++ .../runtime/TrustedSpringBootContextLoader.java | 41 +++++++ .../src/test/resources/application-e2e.yml | 25 +++++ .../src/test/resources/vector.yml | 28 ++--- 15 files changed, 382 insertions(+), 341 deletions(-) diff --git a/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/alert/LogPeriodicAlertE2eTest.java b/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/alert/LogPeriodicAlertE2eTest.java index 9a330a13c3..e175152fe6 100644 --- a/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/alert/LogPeriodicAlertE2eTest.java +++ b/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/alert/LogPeriodicAlertE2eTest.java @@ -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 { - - 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; +public class LogPeriodicAlertE2eTest extends GreptimeE2eSupport { + 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") diff --git a/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/alert/LogRealTimeAlertE2eTest.java b/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/alert/LogRealTimeAlertE2eTest.java index 2af0e6e870..6f79c75184 100644 --- a/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/alert/LogRealTimeAlertE2eTest.java +++ b/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/alert/LogRealTimeAlertE2eTest.java @@ -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(); } diff --git a/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/fixture/GreptimeE2eSupport.java b/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/fixture/GreptimeE2eSupport.java new file mode 100644 index 0000000000..0aea8dbed9 --- /dev/null +++ b/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/fixture/GreptimeE2eSupport.java @@ -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", () -> ""); + } +} diff --git a/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/fixture/VectorE2eContainer.java b/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/fixture/VectorE2eContainer.java new file mode 100644 index 0000000000..1839960c17 --- /dev/null +++ b/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/fixture/VectorE2eContainer.java @@ -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); + } + } +} diff --git a/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/ingestion/LogIngestionE2eTest.java b/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/ingestion/LogIngestionE2eTest.java index 780b233108..be8228380f 100644 --- a/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/ingestion/LogIngestionE2eTest.java +++ b/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/ingestion/LogIngestionE2eTest.java @@ -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(); } diff --git a/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/storage/AuthenticatedGreptimeThreeSignalPublicApiE2eTest.java b/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/storage/AuthenticatedGreptimeThreeSignalPublicApiE2eTest.java index b9af814535..ce681addda 100644 --- a/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/storage/AuthenticatedGreptimeThreeSignalPublicApiE2eTest.java +++ b/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/storage/AuthenticatedGreptimeThreeSignalPublicApiE2eTest.java @@ -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") diff --git a/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/storage/GreptimeLogStorageE2eTest.java b/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/storage/GreptimeLogStorageE2eTest.java index 054dbc5eb4..d23964b79a 100644 --- a/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/storage/GreptimeLogStorageE2eTest.java +++ b/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/storage/GreptimeLogStorageE2eTest.java @@ -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.time.Duration; +import java.util.Collections; 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 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 { - - 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; +public class GreptimeLogStorageE2eTest extends GreptimeE2eSupport { + 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; } diff --git a/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/storage/GreptimeThreeSignalE2eSupport.java b/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/storage/GreptimeThreeSignalE2eSupport.java index bbeb4492ae..aa7fe183fa 100644 --- a/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/storage/GreptimeThreeSignalE2eSupport.java +++ b/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/storage/GreptimeThreeSignalE2eSupport.java @@ -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) { diff --git a/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/storage/GreptimeThreeSignalInstrumentationE2eTest.java b/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/storage/GreptimeThreeSignalInstrumentationE2eTest.java index 6e2b0e07c9..c0a3367c1d 100644 --- a/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/storage/GreptimeThreeSignalInstrumentationE2eTest.java +++ b/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/storage/GreptimeThreeSignalInstrumentationE2eTest.java @@ -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)); diff --git a/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/storage/PrometheusActiveSourcePublicApiE2eTest.java b/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/storage/PrometheusActiveSourcePublicApiE2eTest.java index a9b37eba5e..8c22ee9b3c 100644 --- a/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/storage/PrometheusActiveSourcePublicApiE2eTest.java +++ b/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/storage/PrometheusActiveSourcePublicApiE2eTest.java @@ -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()); diff --git a/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/startup/TrustedStartup.java b/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/startup/TrustedStartup.java new file mode 100644 index 0000000000..bc6cd3f874 --- /dev/null +++ b/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/startup/TrustedStartup.java @@ -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 { +} diff --git a/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/startup/TrustedStartupSupport.java b/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/startup/TrustedStartupSupport.java new file mode 100644 index 0000000000..a672e1dc20 --- /dev/null +++ b/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/startup/TrustedStartupSupport.java @@ -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())); + } + } +} diff --git a/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/startup/runtime/TrustedSpringBootContextLoader.java b/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/startup/runtime/TrustedSpringBootContextLoader.java new file mode 100644 index 0000000000..6506dab3bc --- /dev/null +++ b/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/startup/runtime/TrustedSpringBootContextLoader.java @@ -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); + } + } +} diff --git a/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/resources/application-e2e.yml b/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/resources/application-e2e.yml new file mode 100644 index 0000000000..7cad087c13 --- /dev/null +++ b/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/resources/application-e2e.yml @@ -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 diff --git a/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/resources/vector.yml b/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/resources/vector.yml index d0add7ed93..1fe85eaf94 100644 --- a/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/resources/vector.yml +++ b/hertzbeat-e2e/hertzbeat-observability-e2e/src/test/resources/vector.yml @@ -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 --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
