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 906f4283b69a5c3e5218d15f922aa2b9111cb726 Author: Logic <[email protected]> AuthorDate: Wed Aug 26 15:09:49 2026 +0800 refactor: isolate OTLP guide construction --- .../service/impl/OtlpIngestionGuideFactory.java | 364 +++++++++++++++++++++ .../impl/OtlpIngestionWorkspaceServiceImpl.java | 330 +------------------ .../impl/OtlpIngestionGuideFactoryTest.java | 67 ++++ .../OtlpIngestionWorkspaceServiceImplTest.java | 3 + .../impl/OtlpOverviewWorkspaceIsolationTest.java | 1 + .../impl/OtlpWorkspaceEntityIsolationTest.java | 1 + 6 files changed, 446 insertions(+), 320 deletions(-) diff --git a/hertzbeat-observability/src/main/java/org/apache/hertzbeat/observability/ingestion/service/impl/OtlpIngestionGuideFactory.java b/hertzbeat-observability/src/main/java/org/apache/hertzbeat/observability/ingestion/service/impl/OtlpIngestionGuideFactory.java new file mode 100644 index 0000000000..f93701e0ee --- /dev/null +++ b/hertzbeat-observability/src/main/java/org/apache/hertzbeat/observability/ingestion/service/impl/OtlpIngestionGuideFactory.java @@ -0,0 +1,364 @@ +/* + * 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.ingestion.service.impl; + +import jakarta.servlet.http.HttpServletRequest; +import java.util.List; +import org.apache.hertzbeat.common.observability.dto.ingestion.OtlpIngestionGuideDto; +import org.springframework.beans.factory.annotation.Value; +import org.springframework.stereotype.Component; +import org.springframework.util.StringUtils; + +/** + * Builds the public OTLP endpoints and copy-ready integration snippets. + */ +@Component +class OtlpIngestionGuideFactory { + + private static final String OTLP_HTTP_PATH = "/api/otlp"; + private static final String OTLP_HOST_PLACEHOLDER = "<your-hertzbeat-host>"; + + private final boolean sslEnabled; + private final int serverPort; + private final int otlpGrpcPort; + + OtlpIngestionGuideFactory(@Value("${server.ssl.enabled:false}") boolean sslEnabled, + @Value("${server.port:1157}") int serverPort, + @Value("${hertzbeat.otlp.grpc.port:4317}") int otlpGrpcPort) { + this.sslEnabled = sslEnabled; + this.serverPort = serverPort; + this.otlpGrpcPort = otlpGrpcPort; + } + + OtlpIngestionGuideDto create(HttpServletRequest request) { + String unifiedBaseEndpoint = resolveOtlpHttpBaseEndpoint(request); + String grpcEndpoint = resolveOtlpGrpcEndpoint(request); + String traceEndpoint = unifiedBaseEndpoint + "/v1/traces"; + String metricsEndpoint = unifiedBaseEndpoint + "/v1/metrics"; + String logsEndpoint = unifiedBaseEndpoint + "/v1/logs"; + String collectorSnippet = """ + receivers: + otlp: + protocols: + http: + grpc: + + processors: + batch: + + exporters: + otlphttp/hertzbeat: + endpoint: %s + headers: + Authorization: "Bearer <api-token>" + + service: + pipelines: + logs: + receivers: [otlp] + processors: [batch] + exporters: [otlphttp/hertzbeat] + traces: + receivers: [otlp] + processors: [batch] + exporters: [otlphttp/hertzbeat] + metrics: + receivers: [otlp] + processors: [batch] + exporters: [otlphttp/hertzbeat] + """.formatted(unifiedBaseEndpoint); + String collectorGrpcSnippet = """ + receivers: + otlp: + protocols: + http: + grpc: + + processors: + batch: + + exporters: + otlp/hertzbeat: + endpoint: %s + headers: + Authorization: "Bearer <api-token>" + tls: + insecure: true + + service: + pipelines: + logs: + receivers: [otlp] + processors: [batch] + exporters: [otlp/hertzbeat] + traces: + receivers: [otlp] + processors: [batch] + exporters: [otlp/hertzbeat] + metrics: + receivers: [otlp] + processors: [batch] + exporters: [otlp/hertzbeat] + """.formatted(grpcEndpoint); + String javaSnippet = """ + export OTEL_EXPORTER_OTLP_PROTOCOL=http/protobuf + export OTEL_EXPORTER_OTLP_ENDPOINT=%s + export OTEL_EXPORTER_OTLP_HEADERS=\"Authorization=Bearer <api-token>\" + export OTEL_RESOURCE_ATTRIBUTES=\"service.name=checkout,service.namespace=commerce,deployment.environment.name=prod\" + """.formatted(unifiedBaseEndpoint); + String javaGrpcSnippet = """ + export OTEL_EXPORTER_OTLP_PROTOCOL=grpc + export OTEL_EXPORTER_OTLP_ENDPOINT=%s + export OTEL_EXPORTER_OTLP_HEADERS=\"Authorization=Bearer <api-token>\" + export OTEL_RESOURCE_ATTRIBUTES=\"service.name=checkout,service.namespace=commerce,deployment.environment.name=prod\" + """.formatted("http://"; + grpcEndpoint); + String pythonSnippet = """ + from opentelemetry.sdk.resources import Resource + resource = Resource.create({ + "service.name": "checkout", + "service.namespace": "commerce", + "deployment.environment.name": "prod", + }) + + # %s + # %s + """.formatted(message("observability.otlp.guide.snippet.python.http.comment"), unifiedBaseEndpoint); + String pythonGrpcSnippet = """ + from opentelemetry.sdk.resources import Resource + resource = Resource.create({ + "service.name": "checkout", + "service.namespace": "commerce", + "deployment.environment.name": "prod", + }) + + # %s + # %s + """.formatted(message("observability.otlp.guide.snippet.python.grpc.comment"), grpcEndpoint); + + return new OtlpIngestionGuideDto( + "OTLP HTTP", + "OTLP gRPC", + "Authorization", + "Bearer <api-token>", + grpcEndpoint, + List.of( + signalGuide("metrics", "http", "OTLP HTTP", metricsEndpoint), + signalGuide("logs", "http", "OTLP HTTP", logsEndpoint), + signalGuide("traces", "http", "OTLP HTTP", traceEndpoint), + signalGuide("metrics", "grpc", "OTLP gRPC", grpcEndpoint), + signalGuide("logs", "grpc", "OTLP gRPC", grpcEndpoint), + signalGuide("traces", "grpc", "OTLP gRPC", grpcEndpoint) + ), + List.of( + new OtlpIngestionGuideDto.Snippet( + "collector-http", "http", "OpenTelemetry Collector", "yaml", collectorSnippet), + new OtlpIngestionGuideDto.Snippet( + "java-http", "http", message("observability.otlp.guide.snippet.java-env"), + "bash", javaSnippet), + new OtlpIngestionGuideDto.Snippet( + "python-http", "http", message("observability.otlp.guide.snippet.python-resource"), + "python", pythonSnippet), + new OtlpIngestionGuideDto.Snippet( + "collector-grpc", "grpc", "OpenTelemetry Collector", "yaml", collectorGrpcSnippet), + new OtlpIngestionGuideDto.Snippet( + "java-grpc", "grpc", message("observability.otlp.guide.snippet.java-env"), + "bash", javaGrpcSnippet), + new OtlpIngestionGuideDto.Snippet( + "python-grpc", "grpc", message("observability.otlp.guide.snippet.python-resource"), + "python", pythonGrpcSnippet) + ) + ); + } + + private OtlpIngestionGuideDto.SignalGuide signalGuide(String signal, String protocol, String label, + String endpoint) { + String messagePrefix = "observability.otlp.guide." + signal + "." + protocol; + return new OtlpIngestionGuideDto.SignalGuide( + signal, + protocol, + label, + endpoint, + message(messagePrefix + ".description"), + message(messagePrefix + ".note") + ); + } + + private String resolveOtlpHttpBaseEndpoint(HttpServletRequest request) { + String scheme = firstText( + extractForwardedComponent(request, "proto"), + headerValue(request, "X-Forwarded-Proto"), + sslEnabled ? "https" : "http" + ); + String host = resolveExternalHost(request); + if (!StringUtils.hasText(host)) { + return scheme + "://" + OTLP_HOST_PLACEHOLDER + OTLP_HTTP_PATH; + } + return scheme + "://" + appendPortIfNeeded(host, resolveExternalPort(request, scheme), scheme) + OTLP_HTTP_PATH; + } + + private String resolveOtlpGrpcEndpoint(HttpServletRequest request) { + String host = resolveExternalHost(request); + if (!StringUtils.hasText(host)) { + return OTLP_HOST_PLACEHOLDER + ":" + otlpGrpcPort; + } + return stripPort(host) + ":" + otlpGrpcPort; + } + + private String resolveExternalHost(HttpServletRequest request) { + String forwardedHost = firstText( + extractForwardedComponent(request, "host"), + headerValue(request, "X-Forwarded-Host"), + headerValue(request, "Host") + ); + if (StringUtils.hasText(forwardedHost)) { + return trimForwardedValue(forwardedHost); + } + if (request == null || !StringUtils.hasText(request.getServerName())) { + return null; + } + return request.getServerName(); + } + + private Integer resolveExternalPort(HttpServletRequest request, String scheme) { + String host = resolveExternalHost(request); + String forwardedPort = firstText( + extractForwardedComponent(request, "port"), + headerValue(request, "X-Forwarded-Port") + ); + if (StringUtils.hasText(forwardedPort)) { + try { + return Integer.parseInt(trimForwardedValue(forwardedPort)); + } catch (NumberFormatException ignored) { + // Fall through to the request or configured server port. + } + } + Integer explicitHostPort = extractPortFromHost(host); + if (explicitHostPort != null) { + return explicitHostPort; + } + if (request != null && request.getServerPort() > 0) { + return request.getServerPort(); + } + return "https".equalsIgnoreCase(scheme) ? 443 : serverPort; + } + + private String extractForwardedComponent(HttpServletRequest request, String key) { + String forwarded = headerValue(request, "Forwarded"); + if (!StringUtils.hasText(forwarded)) { + return null; + } + String firstPart = forwarded.split(",", 2)[0]; + for (String token : firstPart.split(";")) { + String[] pair = token.split("=", 2); + if (pair.length == 2 && key.equalsIgnoreCase(pair[0].trim())) { + return trimForwardedValue(pair[1]); + } + } + return null; + } + + private String headerValue(HttpServletRequest request, String headerName) { + return request == null ? null : request.getHeader(headerName); + } + + private String firstText(String... candidates) { + for (String candidate : candidates) { + if (StringUtils.hasText(candidate)) { + return trimForwardedValue(candidate); + } + } + return null; + } + + private String trimForwardedValue(String value) { + if (!StringUtils.hasText(value)) { + return value; + } + String trimmed = value.trim(); + if (trimmed.startsWith("\"") && trimmed.endsWith("\"") && trimmed.length() > 1) { + trimmed = trimmed.substring(1, trimmed.length() - 1); + } + int commaIndex = trimmed.indexOf(','); + if (commaIndex > 0) { + trimmed = trimmed.substring(0, commaIndex); + } + return trimmed.trim(); + } + + private String appendPortIfNeeded(String host, Integer port, String scheme) { + String normalizedHost = stripPort(host); + if (!StringUtils.hasText(normalizedHost) || port == null) { + return normalizedHost; + } + if (("http".equalsIgnoreCase(scheme) && port == 80) + || ("https".equalsIgnoreCase(scheme) && port == 443)) { + return normalizedHost; + } + return normalizedHost + ":" + port; + } + + private String stripPort(String host) { + if (!StringUtils.hasText(host)) { + return host; + } + String trimmed = host.trim(); + if (trimmed.startsWith("[") && trimmed.contains("]")) { + return trimmed.substring(0, trimmed.indexOf(']') + 1); + } + int colonIndex = trimmed.lastIndexOf(':'); + if (colonIndex > 0 && trimmed.indexOf(':') == colonIndex) { + return trimmed.substring(0, colonIndex); + } + if (colonIndex > 0) { + return "[" + trimmed + "]"; + } + return trimmed; + } + + private Integer extractPortFromHost(String host) { + if (!StringUtils.hasText(host)) { + return null; + } + String trimmed = host.trim(); + if (trimmed.startsWith("[") && trimmed.contains("]")) { + int lastColon = trimmed.lastIndexOf(':'); + int closingBracket = trimmed.lastIndexOf(']'); + if (lastColon > closingBracket) { + return parsePort(trimmed.substring(lastColon + 1)); + } + return null; + } + int colonIndex = trimmed.lastIndexOf(':'); + if (colonIndex > 0 && trimmed.indexOf(':') == colonIndex) { + return parsePort(trimmed.substring(colonIndex + 1)); + } + return null; + } + + private Integer parsePort(String value) { + try { + return Integer.parseInt(value); + } catch (NumberFormatException ignored) { + return null; + } + } + + private static String message(String key) { + return OtlpIngestionMessages.get(key); + } +} diff --git a/hertzbeat-observability/src/main/java/org/apache/hertzbeat/observability/ingestion/service/impl/OtlpIngestionWorkspaceServiceImpl.java b/hertzbeat-observability/src/main/java/org/apache/hertzbeat/observability/ingestion/service/impl/OtlpIngestionWorkspaceServiceImpl.java index 69f672bec8..3dcd7e8c7d 100644 --- a/hertzbeat-observability/src/main/java/org/apache/hertzbeat/observability/ingestion/service/impl/OtlpIngestionWorkspaceServiceImpl.java +++ b/hertzbeat-observability/src/main/java/org/apache/hertzbeat/observability/ingestion/service/impl/OtlpIngestionWorkspaceServiceImpl.java @@ -66,7 +66,6 @@ import org.apache.hertzbeat.warehouse.repository.MetricQueryRepository; import org.apache.hertzbeat.warehouse.store.history.tsdb.HistoryDataReader; import org.apache.hertzbeat.warehouse.store.history.tsdb.greptime.GreptimeProperties; import org.springframework.beans.factory.annotation.Qualifier; -import org.springframework.beans.factory.annotation.Value; import org.springframework.stereotype.Service; import org.springframework.util.CollectionUtils; import org.springframework.util.StringUtils; @@ -81,8 +80,6 @@ public class OtlpIngestionWorkspaceServiceImpl implements OtlpIngestionWorkspace private static final long LOOKBACK_MILLIS = Duration.ofHours(24).toMillis(); private static final long DEFAULT_CONSOLE_LOOKBACK_MILLIS = Duration.ofHours(1).toMillis(); private static final int SAMPLE_LIMIT = 20; - private static final String OTLP_HTTP_PATH = "/api/otlp"; - private static final String OTLP_HOST_PLACEHOLDER = "<your-hertzbeat-host>"; private static final List<String> METRICS_ENTITY_CONTEXT_GROUP_LABELS = List.of( "__name__", "service_name", @@ -131,6 +128,7 @@ public class OtlpIngestionWorkspaceServiceImpl implements OtlpIngestionWorkspace private final EntityTraceQueryService entityTraceQueryService; private final ObservabilityWorkspaceQueryGateway workspaceQueryGateway; private final ObservabilitySignalIntakeGateway observabilitySignalIntakeGateway; + private final OtlpIngestionGuideFactory guideFactory; private final LogQueryRepository logQueryRepository; private final MetricQueryRepository metricQueryRepository; private final List<MetricInventoryRepository> metricInventoryRepositories; @@ -142,6 +140,7 @@ public class OtlpIngestionWorkspaceServiceImpl implements OtlpIngestionWorkspace ObservabilityWorkspaceQueryGateway workspaceQueryGateway, @Qualifier("telemetryIntakeServiceImpl") ObservabilitySignalIntakeGateway observabilitySignalIntakeGateway, + OtlpIngestionGuideFactory guideFactory, LogQueryRepository logQueryRepository, MetricQueryRepository metricQueryRepository, List<MetricInventoryRepository> metricInventoryRepositories, @@ -151,6 +150,7 @@ public class OtlpIngestionWorkspaceServiceImpl implements OtlpIngestionWorkspace this.entityTraceQueryService = entityTraceQueryService; this.workspaceQueryGateway = workspaceQueryGateway; this.observabilitySignalIntakeGateway = observabilitySignalIntakeGateway; + this.guideFactory = guideFactory; this.logQueryRepository = logQueryRepository; this.metricQueryRepository = metricQueryRepository; this.metricInventoryRepositories = safeBeanList(metricInventoryRepositories); @@ -159,15 +159,6 @@ public class OtlpIngestionWorkspaceServiceImpl implements OtlpIngestionWorkspace this.greptimeProperties = safeBeanList(greptimeProperties); } - @Value("${server.ssl.enabled:false}") - private boolean sslEnabled = false; - - @Value("${server.port:1157}") - private int serverPort = 1157; - - @Value("${hertzbeat.otlp.grpc.port:4317}") - private int otlpGrpcPort = 4317; - OtlpIngestionOverviewDto getOverview() { return getOverview(AuthTokenScopes.DEFAULT_WORKSPACE_ID); } @@ -436,179 +427,7 @@ public class OtlpIngestionWorkspaceServiceImpl implements OtlpIngestionWorkspace @Override public OtlpIngestionGuideDto getGuide(HttpServletRequest request) { - String unifiedBaseEndpoint = resolveOtlpHttpBaseEndpoint(request); - String grpcEndpoint = resolveOtlpGrpcEndpoint(request); - String traceEndpoint = unifiedBaseEndpoint + "/v1/traces"; - String metricsEndpoint = unifiedBaseEndpoint + "/v1/metrics"; - String logsEndpoint = unifiedBaseEndpoint + "/v1/logs"; - String collectorSnippet = """ - receivers: - otlp: - protocols: - http: - grpc: - - processors: - batch: - - exporters: - otlphttp/hertzbeat: - endpoint: %s - headers: - Authorization: "Bearer <api-token>" - - service: - pipelines: - logs: - receivers: [otlp] - processors: [batch] - exporters: [otlphttp/hertzbeat] - traces: - receivers: [otlp] - processors: [batch] - exporters: [otlphttp/hertzbeat] - metrics: - receivers: [otlp] - processors: [batch] - exporters: [otlphttp/hertzbeat] - """.formatted(unifiedBaseEndpoint); - String collectorGrpcSnippet = """ - receivers: - otlp: - protocols: - http: - grpc: - - processors: - batch: - - exporters: - otlp/hertzbeat: - endpoint: %s - headers: - Authorization: "Bearer <api-token>" - tls: - insecure: true - - service: - pipelines: - logs: - receivers: [otlp] - processors: [batch] - exporters: [otlp/hertzbeat] - traces: - receivers: [otlp] - processors: [batch] - exporters: [otlp/hertzbeat] - metrics: - receivers: [otlp] - processors: [batch] - exporters: [otlp/hertzbeat] - """.formatted(grpcEndpoint); - String javaSnippet = """ - export OTEL_EXPORTER_OTLP_PROTOCOL=http/protobuf - export OTEL_EXPORTER_OTLP_ENDPOINT=%s - export OTEL_EXPORTER_OTLP_HEADERS=\"Authorization=Bearer <api-token>\" - export OTEL_RESOURCE_ATTRIBUTES=\"service.name=checkout,service.namespace=commerce,deployment.environment.name=prod\" - """.formatted(unifiedBaseEndpoint); - String javaGrpcSnippet = """ - export OTEL_EXPORTER_OTLP_PROTOCOL=grpc - export OTEL_EXPORTER_OTLP_ENDPOINT=%s - export OTEL_EXPORTER_OTLP_HEADERS=\"Authorization=Bearer <api-token>\" - export OTEL_RESOURCE_ATTRIBUTES=\"service.name=checkout,service.namespace=commerce,deployment.environment.name=prod\" - """.formatted("http://"; + grpcEndpoint); - String pythonSnippet = """ - from opentelemetry.sdk.resources import Resource - resource = Resource.create({ - "service.name": "checkout", - "service.namespace": "commerce", - "deployment.environment.name": "prod", - }) - - # %s - # %s - """.formatted(message("observability.otlp.guide.snippet.python.http.comment"), unifiedBaseEndpoint); - String pythonGrpcSnippet = """ - from opentelemetry.sdk.resources import Resource - resource = Resource.create({ - "service.name": "checkout", - "service.namespace": "commerce", - "deployment.environment.name": "prod", - }) - - # %s - # %s - """.formatted(message("observability.otlp.guide.snippet.python.grpc.comment"), grpcEndpoint); - - return new OtlpIngestionGuideDto( - "OTLP HTTP", - "OTLP gRPC", - "Authorization", - "Bearer <api-token>", - grpcEndpoint, - List.of( - new OtlpIngestionGuideDto.SignalGuide( - "metrics", - "http", - "OTLP HTTP", - metricsEndpoint, - message("observability.otlp.guide.metrics.http.description"), - message("observability.otlp.guide.metrics.http.note") - ), - new OtlpIngestionGuideDto.SignalGuide( - "logs", - "http", - "OTLP HTTP", - logsEndpoint, - message("observability.otlp.guide.logs.http.description"), - message("observability.otlp.guide.logs.http.note") - ), - new OtlpIngestionGuideDto.SignalGuide( - "traces", - "http", - "OTLP HTTP", - traceEndpoint, - message("observability.otlp.guide.traces.http.description"), - message("observability.otlp.guide.traces.http.note") - ), - new OtlpIngestionGuideDto.SignalGuide( - "metrics", - "grpc", - "OTLP gRPC", - grpcEndpoint, - message("observability.otlp.guide.metrics.grpc.description"), - message("observability.otlp.guide.metrics.grpc.note") - ), - new OtlpIngestionGuideDto.SignalGuide( - "logs", - "grpc", - "OTLP gRPC", - grpcEndpoint, - message("observability.otlp.guide.logs.grpc.description"), - message("observability.otlp.guide.logs.grpc.note") - ), - new OtlpIngestionGuideDto.SignalGuide( - "traces", - "grpc", - "OTLP gRPC", - grpcEndpoint, - message("observability.otlp.guide.traces.grpc.description"), - message("observability.otlp.guide.traces.grpc.note") - ) - ), - List.of( - new OtlpIngestionGuideDto.Snippet("collector-http", "http", "OpenTelemetry Collector", "yaml", collectorSnippet), - new OtlpIngestionGuideDto.Snippet("java-http", "http", - message("observability.otlp.guide.snippet.java-env"), "bash", javaSnippet), - new OtlpIngestionGuideDto.Snippet("python-http", "http", - message("observability.otlp.guide.snippet.python-resource"), "python", pythonSnippet), - new OtlpIngestionGuideDto.Snippet("collector-grpc", "grpc", "OpenTelemetry Collector", "yaml", collectorGrpcSnippet), - new OtlpIngestionGuideDto.Snippet("java-grpc", "grpc", - message("observability.otlp.guide.snippet.java-env"), "bash", javaGrpcSnippet), - new OtlpIngestionGuideDto.Snippet("python-grpc", "grpc", - message("observability.otlp.guide.snippet.python-resource"), "python", pythonGrpcSnippet) - ) - ); + return guideFactory.create(request); } OtlpEntityBindingSummaryDto getBindingSummary() { @@ -1758,87 +1577,6 @@ public class OtlpIngestionWorkspaceServiceImpl implements OtlpIngestionWorkspace return localDateTime.atZone(ZoneId.systemDefault()).toInstant().toEpochMilli(); } - private String resolveOtlpHttpBaseEndpoint(HttpServletRequest request) { - String scheme = firstText( - extractForwardedComponent(request, "proto"), - headerValue(request, "X-Forwarded-Proto"), - sslEnabled ? "https" : "http" - ); - String host = resolveExternalHost(request); - if (!StringUtils.hasText(host)) { - return scheme + "://" + OTLP_HOST_PLACEHOLDER + OTLP_HTTP_PATH; - } - return scheme + "://" + appendPortIfNeeded(host, resolveExternalPort(request, scheme), scheme) + OTLP_HTTP_PATH; - } - - private String resolveOtlpGrpcEndpoint(HttpServletRequest request) { - String host = resolveExternalHost(request); - if (!StringUtils.hasText(host)) { - return OTLP_HOST_PLACEHOLDER + ":" + otlpGrpcPort; - } - return stripPort(host) + ":" + otlpGrpcPort; - } - - private String resolveExternalHost(HttpServletRequest request) { - String forwardedHost = firstText( - extractForwardedComponent(request, "host"), - headerValue(request, "X-Forwarded-Host"), - headerValue(request, "Host") - ); - if (StringUtils.hasText(forwardedHost)) { - return trimForwardedValue(forwardedHost); - } - if (request == null || !StringUtils.hasText(request.getServerName())) { - return null; - } - return request.getServerName(); - } - - private Integer resolveExternalPort(HttpServletRequest request, String scheme) { - String host = resolveExternalHost(request); - String forwardedPort = firstText( - extractForwardedComponent(request, "port"), - headerValue(request, "X-Forwarded-Port") - ); - if (StringUtils.hasText(forwardedPort)) { - try { - return Integer.parseInt(trimForwardedValue(forwardedPort)); - } catch (NumberFormatException ignored) { - // Ignore invalid forwarded port and fall back to request or server defaults. - } - } - Integer explicitHostPort = extractPortFromHost(host); - if (explicitHostPort != null) { - return explicitHostPort; - } - if (request != null && request.getServerPort() > 0) { - return request.getServerPort(); - } - return "https".equalsIgnoreCase(scheme) ? 443 : serverPort; - } - - private String extractForwardedComponent(HttpServletRequest request, String key) { - String forwarded = headerValue(request, "Forwarded"); - if (!StringUtils.hasText(forwarded)) { - return null; - } - String firstPart = forwarded.split(",", 2)[0]; - for (String token : firstPart.split(";")) { - String[] pair = token.split("=", 2); - if (pair.length == 2 && key.equalsIgnoreCase(pair[0].trim())) { - return trimForwardedValue(pair[1]); - } - } - return null; - } - - private String headerValue(HttpServletRequest request, String headerName) { - if (request == null) { - return null; - } - return request.getHeader(headerName); - } - private String firstText(String... candidates) { for (String candidate : candidates) { if (StringUtils.hasText(candidate)) { @@ -1863,62 +1601,14 @@ public class OtlpIngestionWorkspaceServiceImpl implements OtlpIngestionWorkspace return trimmed.trim(); } - private String appendPortIfNeeded(String host, Integer port, String scheme) { - String normalizedHost = stripPort(host); - if (!StringUtils.hasText(normalizedHost) || port == null) { - return normalizedHost; - } - if (("http".equalsIgnoreCase(scheme) && port == 80) - || ("https".equalsIgnoreCase(scheme) && port == 443)) { - return normalizedHost; - } - return normalizedHost + ":" + port; - } - - private String stripPort(String host) { - if (!StringUtils.hasText(host)) { - return host; - } - String trimmed = host.trim(); - if (trimmed.startsWith("[") && trimmed.contains("]")) { - return trimmed.substring(0, trimmed.indexOf(']') + 1); - } - int colonIndex = trimmed.lastIndexOf(':'); - if (colonIndex > 0 && trimmed.indexOf(':') == colonIndex) { - return trimmed.substring(0, colonIndex); + private String metricsIntakeMode(long monitorTotalCount, long otlpMetricCount) { + if (monitorTotalCount > 0 && otlpMetricCount > 0) { + return message("observability.otlp.overview.metrics.mode.mixed"); } - if (colonIndex > 0 && trimmed.indexOf(':') != colonIndex) { - return "[" + trimmed + "]"; + if (otlpMetricCount > 0) { + return "OTLP"; } - return trimmed; - } - - private Integer extractPortFromHost(String host) { - if (!StringUtils.hasText(host)) { - return null; - } - String trimmed = host.trim(); - if (trimmed.startsWith("[") && trimmed.contains("]")) { - int lastColon = trimmed.lastIndexOf(':'); - int closingBracket = trimmed.lastIndexOf(']'); - if (lastColon > closingBracket) { - try { - return Integer.parseInt(trimmed.substring(lastColon + 1)); - } catch (NumberFormatException ignored) { - return null; - } - } - return null; - } - int colonIndex = trimmed.lastIndexOf(':'); - if (colonIndex > 0 && trimmed.indexOf(':') == colonIndex) { - try { - return Integer.parseInt(trimmed.substring(colonIndex + 1)); - } catch (NumberFormatException ignored) { - return null; - } - } - return null; + return message("observability.otlp.overview.metrics.mode.monitor"); } private static String message(String key) { diff --git a/hertzbeat-observability/src/test/java/org/apache/hertzbeat/observability/ingestion/service/impl/OtlpIngestionGuideFactoryTest.java b/hertzbeat-observability/src/test/java/org/apache/hertzbeat/observability/ingestion/service/impl/OtlpIngestionGuideFactoryTest.java new file mode 100644 index 0000000000..50efaa0d4c --- /dev/null +++ b/hertzbeat-observability/src/test/java/org/apache/hertzbeat/observability/ingestion/service/impl/OtlpIngestionGuideFactoryTest.java @@ -0,0 +1,67 @@ +/* + * 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.ingestion.service.impl; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertTrue; + +import org.apache.hertzbeat.common.observability.dto.ingestion.OtlpIngestionGuideDto; +import org.junit.jupiter.api.Test; +import org.springframework.mock.web.MockHttpServletRequest; + +class OtlpIngestionGuideFactoryTest { + + private final OtlpIngestionGuideFactory factory = new OtlpIngestionGuideFactory(false, 1157, 4317); + + @Test + void buildsHttpAndGrpcEndpointsFromForwardedHeaders() { + MockHttpServletRequest request = new MockHttpServletRequest(); + request.setServerPort(443); + request.addHeader("X-Forwarded-Host", "demo.hertzbeat.apache.org"); + request.addHeader("X-Forwarded-Proto", "https"); + + OtlpIngestionGuideDto guide = factory.create(request); + + assertEquals("demo.hertzbeat.apache.org:4317", guide.getGrpcAuthorityExample()); + assertTrue(guide.getSignals().stream().anyMatch(signal -> "metrics".equals(signal.getSignal()) + && "https://demo.hertzbeat.apache.org/api/otlp/v1/metrics".equals(signal.getEndpoint()))); + } + + @Test + void preservesBracketedIpv6AndExplicitForwardedPort() { + MockHttpServletRequest request = new MockHttpServletRequest(); + request.addHeader("Forwarded", "proto=https;host=\"[2001:db8::1]:8443\""); + + OtlpIngestionGuideDto guide = factory.create(request); + + assertEquals("[2001:db8::1]:4317", guide.getGrpcAuthorityExample()); + assertTrue(guide.getSignals().stream().anyMatch(signal -> "logs".equals(signal.getSignal()) + && "https://[2001:db8::1]:8443/api/otlp/v1/logs".equals(signal.getEndpoint()))); + } + + @Test + void usesConfiguredFallbackEndpointsWhenRequestHostIsMissing() { + OtlpIngestionGuideFactory secureFactory = new OtlpIngestionGuideFactory(true, 8443, 55680); + + OtlpIngestionGuideDto guide = secureFactory.create(null); + + assertEquals("<your-hertzbeat-host>:55680", guide.getGrpcAuthorityExample()); + assertTrue(guide.getSignals().stream().anyMatch(signal -> "traces".equals(signal.getSignal()) + && "https://<your-hertzbeat-host>/api/otlp/v1/traces".equals(signal.getEndpoint()))); + } +} diff --git a/hertzbeat-observability/src/test/java/org/apache/hertzbeat/observability/ingestion/service/impl/OtlpIngestionWorkspaceServiceImplTest.java b/hertzbeat-observability/src/test/java/org/apache/hertzbeat/observability/ingestion/service/impl/OtlpIngestionWorkspaceServiceImplTest.java index 7f1f024e0f..9cbe81d56d 100644 --- a/hertzbeat-observability/src/test/java/org/apache/hertzbeat/observability/ingestion/service/impl/OtlpIngestionWorkspaceServiceImplTest.java +++ b/hertzbeat-observability/src/test/java/org/apache/hertzbeat/observability/ingestion/service/impl/OtlpIngestionWorkspaceServiceImplTest.java @@ -118,6 +118,7 @@ class OtlpIngestionWorkspaceServiceImplTest { entityTraceQueryService, workspaceQueryGateway, observabilitySignalIntakeGateway, + new OtlpIngestionGuideFactory(false, 1157, 4317), logQueryRepository, metricQueryRepository, List.of(metricInventoryRepository), @@ -313,6 +314,7 @@ class OtlpIngestionWorkspaceServiceImplTest { entityTraceQueryService, workspaceQueryGateway, observabilitySignalIntakeGateway, + new OtlpIngestionGuideFactory(false, 1157, 4317), logQueryRepository, metricQueryRepository, List.of(metricInventoryRepository), @@ -634,6 +636,7 @@ class OtlpIngestionWorkspaceServiceImplTest { entityTraceQueryService, workspaceQueryGateway, mixedSourceGateway, + new OtlpIngestionGuideFactory(false, 1157, 4317), logQueryRepository, metricQueryRepository, List.of(metricInventoryRepository), diff --git a/hertzbeat-observability/src/test/java/org/apache/hertzbeat/observability/ingestion/service/impl/OtlpOverviewWorkspaceIsolationTest.java b/hertzbeat-observability/src/test/java/org/apache/hertzbeat/observability/ingestion/service/impl/OtlpOverviewWorkspaceIsolationTest.java index ff1729cc91..5a45cc9fdf 100644 --- a/hertzbeat-observability/src/test/java/org/apache/hertzbeat/observability/ingestion/service/impl/OtlpOverviewWorkspaceIsolationTest.java +++ b/hertzbeat-observability/src/test/java/org/apache/hertzbeat/observability/ingestion/service/impl/OtlpOverviewWorkspaceIsolationTest.java @@ -67,6 +67,7 @@ class OtlpOverviewWorkspaceIsolationTest { traceQueryService, workspaceQueryGateway, intakeGateway, + new OtlpIngestionGuideFactory(false, 1157, 4317), logQueryRepository, metricQueryRepository, List.of(metricInventoryRepository), diff --git a/hertzbeat-observability/src/test/java/org/apache/hertzbeat/observability/ingestion/service/impl/OtlpWorkspaceEntityIsolationTest.java b/hertzbeat-observability/src/test/java/org/apache/hertzbeat/observability/ingestion/service/impl/OtlpWorkspaceEntityIsolationTest.java index 85ab2be12a..d68e3d28a1 100644 --- a/hertzbeat-observability/src/test/java/org/apache/hertzbeat/observability/ingestion/service/impl/OtlpWorkspaceEntityIsolationTest.java +++ b/hertzbeat-observability/src/test/java/org/apache/hertzbeat/observability/ingestion/service/impl/OtlpWorkspaceEntityIsolationTest.java @@ -62,6 +62,7 @@ class OtlpWorkspaceEntityIsolationTest { entityTraceQueryService, workspaceQueryGateway, signalIntakeGateway, + new OtlpIngestionGuideFactory(false, 1157, 4317), logQueryRepository, metricQueryRepository, List.of(metricInventoryRepository), --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
