Consolidate entity-free observability for 1.9

.github/workflows/backend-build-test.yml

@@ -76,7 +76,7 @@ jobs:
       - name: Build backend Maven E2E modules
         run: |
           mvnd clean -B package \
-            -pl hertzbeat-e2e/hertzbeat-collector-common-e2e,hertzbeat-e2e/hertzbeat-collector-kafka-e2e,hertzbeat-e2e/hertzbeat-collector-basic-e2e,hertzbeat-e2e/hertzbeat-collector-mysql-r2dbc-e2e,hertzbeat-e2e/hertzbeat-log-e2e \
+            -pl hertzbeat-e2e/hertzbeat-collector-common-e2e,hertzbeat-e2e/hertzbeat-collector-kafka-e2e,hertzbeat-e2e/hertzbeat-collector-basic-e2e,hertzbeat-e2e/hertzbeat-collector-mysql-r2dbc-e2e,hertzbeat-e2e/hertzbeat-observability-e2e \
             -am \
             -Dmaven.test.skip=false \
             --file pom.xml

hertzbeat-common-core/src/main/java/org/apache/hertzbeat/common/constants/ConfigConstants.java

@@ -69,6 +69,8 @@ interface FunctionModuleConstants {
         String GRAFANA = "grafana";
 
         String LOG = "log";
+
+        String OBSERVABILITY = "observability";
     }
 
 }

hertzbeat-common-core/src/main/java/org/apache/hertzbeat/common/constants/NetworkConstants.java

@@ -61,7 +61,7 @@ interface HttpClientConstants {
         Duration READ_TIMEOUT = Duration.ofSeconds(6);
         Duration WRITE_TIMEOUT = Duration.ofSeconds(6);
         Duration CONNECT_TIMEOUT = Duration.ofSeconds(6);
-        Duration GREPTIME_QUERY_READ_TIMEOUT = Duration.ofSeconds(5);
+        Duration GREPTIME_QUERY_READ_TIMEOUT = Duration.ofSeconds(15);
         Duration GREPTIME_QUERY_CONNECT_TIMEOUT = Duration.ofSeconds(2);
         Duration GREPTIME_WRITE_READ_TIMEOUT = Duration.ofSeconds(3);
         Duration GREPTIME_WRITE_CONNECT_TIMEOUT = Duration.ofSeconds(2);

hertzbeat-e2e/hertzbeat-log-e2e/pom.xml → hertzbeat-e2e/hertzbeat-observability-e2e/pom.xml

@@ -25,7 +25,7 @@
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
-    <artifactId>hertzbeat-log-e2e</artifactId>
+    <artifactId>hertzbeat-observability-e2e</artifactId>
 
     <properties>
         <maven.compiler.source>${java.version}</maven.compiler.source>
@@ -52,7 +52,7 @@
         </dependency>
         <dependency>
             <groupId>org.apache.hertzbeat</groupId>
-            <artifactId>hertzbeat-log</artifactId>
+            <artifactId>hertzbeat-observability</artifactId>
             <version>${hertzbeat.version}</version>
             <scope>test</scope>
         </dependency>

hertzbeat-e2e/hertzbeat-log-e2e/src/test/java/org/apache/hertzbeat/log/alert/LogPeriodicAlertE2eTest.java → hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/alert/LogPeriodicAlertE2eTest.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.hertzbeat.log.alert;
+package org.apache.hertzbeat.observability.alert;
 
 import lombok.extern.slf4j.Slf4j;
 import org.apache.hertzbeat.alert.calculate.periodic.PeriodicAlertRuleScheduler;

hertzbeat-e2e/hertzbeat-log-e2e/src/test/java/org/apache/hertzbeat/log/alert/LogRealTimeAlertE2eTest.java → hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/alert/LogRealTimeAlertE2eTest.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.hertzbeat.log.alert;
+package org.apache.hertzbeat.observability.alert;
 
 import lombok.extern.slf4j.Slf4j;
 import org.apache.hertzbeat.alert.reduce.AlarmCommonReduce;

hertzbeat-e2e/hertzbeat-log-e2e/src/test/java/org/apache/hertzbeat/log/ingestion/LogIngestionE2eTest.java → hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/ingestion/LogIngestionE2eTest.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.hertzbeat.log.ingestion;
+package org.apache.hertzbeat.observability.ingestion;
 
 import lombok.extern.slf4j.Slf4j;
 import org.apache.hertzbeat.common.entity.log.LogEntry;

hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/storage/GreptimeEntityFreeSignalStorageE2eTest.java

@@ -0,0 +1,176 @@
+/*
+ * 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
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hertzbeat.observability.storage;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.awaitility.Awaitility.await;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.protobuf.ByteString;
+import io.opentelemetry.proto.collector.logs.v1.ExportLogsServiceRequest;
+import io.opentelemetry.proto.common.v1.AnyValue;
+import io.opentelemetry.proto.common.v1.KeyValue;
+import io.opentelemetry.proto.logs.v1.LogRecord;
+import io.opentelemetry.proto.logs.v1.ResourceLogs;
+import io.opentelemetry.proto.logs.v1.ScopeLogs;
+import io.opentelemetry.proto.resource.v1.Resource;
+import java.io.InputStream;
+import java.net.URI;
+import java.net.URLEncoder;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
+import java.nio.charset.StandardCharsets;
+import java.time.Duration;
+import org.apache.hertzbeat.warehouse.store.history.tsdb.greptime.GreptimeOtlpSignalStorage;
+import org.apache.hertzbeat.warehouse.store.history.tsdb.greptime.GreptimeProperties;
+import org.junit.jupiter.api.Test;
+import org.springframework.web.client.RestTemplate;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.junit.jupiter.Container;
+import org.testcontainers.junit.jupiter.Testcontainers;
+import org.testcontainers.utility.DockerImageName;
+
+/** Proves the warehouse-owned Entity-free OTLP log path against a real GreptimeDB. */
+@Testcontainers
+class GreptimeEntityFreeSignalStorageE2eTest {
+
+    private static final int GREPTIME_HTTP_PORT = 4000;
+    private static final int GREPTIME_GRPC_PORT = 4001;
+    private static final String LOG_SCHEMA = "greptime/tables/hertzbeat_logs.sql";
+    private static final String LOG_PIPELINE = "greptime/pipelines/hertzbeat_otlp_log_v1.yaml";
+    private static final String PIPELINE_NAME = "hertzbeat_otlp_log_v1";
+    private static final String TRACE_ID = "0123456789abcdef0123456789abcdef";
+    private static final String SPAN_ID = "0123456789abcdef";
+    private static final String BODY = "entity-free greptime proof";
+    private static final long LOG_TIME_NANOS = 1_710_000_000_123_456_789L;
+    private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+    @Container
+    @SuppressWarnings("resource")
+    private static final GenericContainer<?> GREPTIME = new GenericContainer<>(
+            DockerImageName.parse("greptime/greptimedb:latest"))
+            .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));
+
+    private final HttpClient httpClient = HttpClient.newHttpClient();
+
+    @Test
+    void warehouseStorageShouldPersistEntityFreeOtlpLogs() throws Exception {
+        executeSql(classpathResource(LOG_SCHEMA).strip().replaceFirst(";\\s*$", ""));
+        uploadPipeline();
+
+        GreptimeOtlpSignalStorage storage = new GreptimeOtlpSignalStorage(
+                new GreptimeProperties(true, GREPTIME.getHost() + ':' + GREPTIME.getMappedPort(GREPTIME_GRPC_PORT),
+                        endpoint(), "public", "", ""),
+                new RestTemplate());
+
+        storage.writeProtobuf("logs", request().toByteArray());
+
+        await().atMost(Duration.ofSeconds(30)).pollInterval(Duration.ofSeconds(1)).untilAsserted(() -> {
+            String sql = "SELECT COUNT(*) AS count FROM hertzbeat_logs WHERE trace_id = '" + TRACE_ID
+                    + "' AND body = '" + BODY + "'";
+            assertThat(queryCount(sql)).isEqualTo(1);
+        });
+    }
+
+    private void uploadPipeline() throws Exception {
+        String boundary = "----hertzbeat-entity-free-proof";
+        String body = "--" + boundary + "\r\n"
+                + "Content-Disposition: form-data; name=\"file\"; filename=\"pipeline.yaml\"\r\n"
+                + "Content-Type: application/x-yaml\r\n\r\n"
+                + classpathResource(LOG_PIPELINE) + "\r\n"
+                + "--" + boundary + "--\r\n";
+        HttpResponse<String> response = httpClient.send(HttpRequest.newBuilder()
+                .uri(URI.create(endpoint() + "/v1/pipelines/" + PIPELINE_NAME))
+                .header("Content-Type", "multipart/form-data; boundary=" + boundary)
+                .POST(HttpRequest.BodyPublishers.ofString(body, StandardCharsets.UTF_8))
+                .build(), HttpResponse.BodyHandlers.ofString());
+        assertThat(response.statusCode()).as(response.body()).isBetween(200, 299);
+    }
+
+    private ExportLogsServiceRequest request() {
+        LogRecord record = LogRecord.newBuilder()
+                .setTimeUnixNano(LOG_TIME_NANOS)
+                .setObservedTimeUnixNano(LOG_TIME_NANOS)
+                .setSeverityNumberValue(9)
+                .setSeverityText("INFO")
+                .setBody(AnyValue.newBuilder().setStringValue(BODY).build())
+                .setTraceId(ByteString.copyFrom(hexToBytes(TRACE_ID)))
+                .setSpanId(ByteString.copyFrom(hexToBytes(SPAN_ID)))
+                .build();
+        return ExportLogsServiceRequest.newBuilder()
+                .addResourceLogs(ResourceLogs.newBuilder()
+                        .setResource(Resource.newBuilder()
+                                .addAttributes(stringAttribute("service.name", "checkout"))
+                                .addAttributes(stringAttribute("deployment.environment.name", "test"))
+                                .build())
+                        .addScopeLogs(ScopeLogs.newBuilder().addLogRecords(record).build())
+                        .build())
+                .build();
+    }
+
+    private int queryCount(String sql) throws Exception {
+        JsonNode rows = OBJECT_MAPPER.readTree(executeSql(sql).body())
+                .path("output").path(0).path("records").path("rows");
+        assertThat(rows.isArray()).isTrue();
+        assertThat(rows).isNotEmpty();
+        return rows.get(0).get(0).asInt();
+    }
+
+    private HttpResponse<String> executeSql(String sql) throws Exception {
+        HttpResponse<String> response = httpClient.send(HttpRequest.newBuilder()
+                .uri(URI.create(endpoint() + "/v1/sql?db=public"))
+                .header("Content-Type", "application/x-www-form-urlencoded")
+                .POST(HttpRequest.BodyPublishers.ofString(
+                        "sql=" + URLEncoder.encode(sql, StandardCharsets.UTF_8), StandardCharsets.UTF_8))
+                .build(), HttpResponse.BodyHandlers.ofString());
+        assertThat(response.statusCode()).as(response.body()).isBetween(200, 299);
+        return response;
+    }
+
+    private String classpathResource(String path) throws Exception {
+        try (InputStream input = Thread.currentThread().getContextClassLoader().getResourceAsStream(path)) {
+            assertThat(input).as(path).isNotNull();
+            return new String(input.readAllBytes(), StandardCharsets.UTF_8);
+        }
+    }
+
+    private static KeyValue stringAttribute(String key, String value) {
+        return KeyValue.newBuilder().setKey(key)
+                .setValue(AnyValue.newBuilder().setStringValue(value).build()).build();
+    }
+
+    private static byte[] hexToBytes(String value) {
+        byte[] bytes = new byte[value.length() / 2];
+        for (int index = 0; index < value.length(); index += 2) {
+            bytes[index / 2] = (byte) Integer.parseInt(value.substring(index, index + 2), 16);
+        }
+        return bytes;
+    }
+
+    private static String endpoint() {
+        return "http://" + GREPTIME.getHost() + ':' + GREPTIME.getMappedPort(GREPTIME_HTTP_PORT);
+    }
+}

hertzbeat-e2e/hertzbeat-log-e2e/src/test/java/org/apache/hertzbeat/log/storage/GreptimeLogStorageE2eTest.java → hertzbeat-e2e/hertzbeat-observability-e2e/src/test/java/org/apache/hertzbeat/observability/storage/GreptimeLogStorageE2eTest.java

@@ -15,7 +15,7 @@
  * limitations under the License.
  */
 
-package org.apache.hertzbeat.log.storage;
+package org.apache.hertzbeat.observability.storage;
 
 import lombok.extern.slf4j.Slf4j;
 import org.apache.hertzbeat.common.entity.log.LogEntry;

hertzbeat-e2e/hertzbeat-log-e2e/src/test/resources/sureness.yml → hertzbeat-e2e/hertzbeat-observability-e2e/src/test/resources/sureness.yml

@@ -74,12 +74,9 @@ resourceRole:
   - /api/ai/**===post===[admin]
   - /api/ai/**===put===[admin]
   - /api/ai/**===delete===[admin]
-  - /api/logs/sse/**===get===[admin,user,guest]
-  - /api/logs/ingest/**===post===[admin,user]
-  - /api/otlp/**===post===[admin,user]
-  - /api/ingestion/otlp/**===get===[admin,user,guest]
-  - /api/logs/**===get===[admin,user,guest]
-  - /api/traces/**===get===[admin,user,guest]
+  - /api/otlp/v1/**===post===[admin,user]
+  - /api/observability/logs===delete===[admin]
+  - /api/observability/**===get===[admin,user,guest]
 
 # config the resource restful api that need bypass auth protection
 # rule: api===method 
@@ -89,7 +86,6 @@ excludedResource:
   - /api/account/auth/**===*
   - /api/i18n/**===get
   - /api/apps/hierarchy===get
-  - /api/observability/capability===get
   - /api/push/**===*
   - /api/status/page/public/**===*
   - /api/manager/sse/**===*

hertzbeat-e2e/hertzbeat-log-e2e/src/test/resources/vector.yml → hertzbeat-e2e/hertzbeat-observability-e2e/src/test/resources/vector.yml

@@ -102,7 +102,7 @@ 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:-1157}/api/otlp/v1/logs"
       method: post
       encoding:
         codec: json

hertzbeat-e2e/pom.xml

@@ -32,7 +32,7 @@
         <module>hertzbeat-collector-kafka-e2e</module>
         <module>hertzbeat-collector-basic-e2e</module>
         <module>hertzbeat-collector-mysql-r2dbc-e2e</module>
-        <module>hertzbeat-log-e2e</module>
+        <module>hertzbeat-observability-e2e</module>
     </modules>
 
     <properties>