From a782fc106a552e31db1c2e01091f745dadc3329b Mon Sep 17 00:00:00 2001 From: mrproliu <741550557@qq.com> Date: Tue, 13 May 2025 17:01:05 +0800 Subject: [PATCH] Support analysis waypoint metrics in Envoy ALS receiver (#13244) --- docs/en/changes/changes.md | 1 + .../receiver/envoy/als/AccessLogAnalyzer.java | 2 + .../oap/server/receiver/envoy/als/Role.java | 6 +- .../receiver/envoy/als/k8s/Addresses.java | 29 ++++++- .../k8s/K8sALSServiceMeshHTTPAnalysis.java | 80 ++++++++++++++++--- .../proto/envoy/config/core/v3/address.proto | 51 +++++++----- 6 files changed, 132 insertions(+), 37 deletions(-) diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md index 8dfce58905..8c83afb1b8 100644 --- a/docs/en/changes/changes.md +++ b/docs/en/changes/changes.md @@ -16,6 +16,7 @@ * BanyanDB: Support cold stage data query for metrics/traces/logs. * Increase the idle check interval of the message queue to 200ms to reduce CPU usage under low load conditions. * Limit max attempts of DNS resolution of Istio ServiceEntry to 3, and do not wait for first resolution result in case the DNS is not resolvable at all. +* Support analysis waypoint metrics in Envoy ALS receiver. #### UI diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/AccessLogAnalyzer.java b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/AccessLogAnalyzer.java index 187f023213..c144d18ec6 100644 --- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/AccessLogAnalyzer.java +++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/AccessLogAnalyzer.java @@ -65,6 +65,8 @@ public interface AccessLogAnalyzer { return Role.PROXY; } else if (id.startsWith("sidecar~")) { return Role.SIDECAR; + } else if (id.startsWith("waypoint~")) { + return Role.WAYPOINT; } return defaultRole; } diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/Role.java b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/Role.java index 8842c39b8b..3a196b3f35 100644 --- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/Role.java +++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/Role.java @@ -33,5 +33,9 @@ public enum Role { /** * Sidecar in mesh */ - SIDECAR + SIDECAR, + /** + * Waypoint in ambient mesh + */ + WAYPOINT, } diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/k8s/Addresses.java b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/k8s/Addresses.java index 1b5fc35cfb..8b8ee291a3 100644 --- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/k8s/Addresses.java +++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/k8s/Addresses.java @@ -20,13 +20,34 @@ package org.apache.skywalking.oap.server.receiver.envoy.als.k8s; import io.envoyproxy.envoy.config.core.v3.Address; -import static java.util.Objects.nonNull; +import static java.util.Objects.isNull; import static org.apache.skywalking.oap.server.library.util.StringUtil.isNotBlank; public class Addresses { public static boolean isValid(final Address address) { - return nonNull(address) - && address.hasSocketAddress() - && isNotBlank(address.getSocketAddress().getAddress()); + if (isNull(address)) { + return false; + } + if (address.hasSocketAddress()) { + return isNotBlank(address.getSocketAddress().getAddress()); + } + if (address.hasEnvoyInternalAddress()) { + return isNotBlank(address.getEnvoyInternalAddress().getEndpointId()) && + address.getEnvoyInternalAddress().getEndpointId().split(":").length == 2; + } + return false; + } + + public static String getAddressIP(final Address address) { + if (isNull(address)) { + return null; + } + if (address.hasSocketAddress()) { + return address.getSocketAddress().getAddress(); + } + if (address.hasEnvoyInternalAddress()) { + return address.getEnvoyInternalAddress().getEndpointId().split(":")[0]; + } + return null; } } diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/k8s/K8sALSServiceMeshHTTPAnalysis.java b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/k8s/K8sALSServiceMeshHTTPAnalysis.java index e4b0368f55..ee32d170a1 100644 --- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/k8s/K8sALSServiceMeshHTTPAnalysis.java +++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/k8s/K8sALSServiceMeshHTTPAnalysis.java @@ -18,8 +18,8 @@ package org.apache.skywalking.oap.server.receiver.envoy.als.k8s; +import com.google.protobuf.Value; import io.envoyproxy.envoy.config.core.v3.Address; -import io.envoyproxy.envoy.config.core.v3.SocketAddress; import io.envoyproxy.envoy.data.accesslog.v3.AccessLogCommon; import io.envoyproxy.envoy.data.accesslog.v3.HTTPAccessLogEntry; import io.envoyproxy.envoy.service.accesslog.v3.StreamAccessLogsMessage; @@ -28,6 +28,7 @@ import lombok.extern.slf4j.Slf4j; import org.apache.skywalking.apm.network.servicemesh.v3.HTTPServiceMeshMetric; import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetrics; import org.apache.skywalking.oap.server.library.module.ModuleManager; +import org.apache.skywalking.oap.server.library.util.StringUtil; import org.apache.skywalking.oap.server.receiver.envoy.EnvoyMetricReceiverConfig; import org.apache.skywalking.oap.server.receiver.envoy.ServiceMetaInfoFactory; import org.apache.skywalking.oap.server.receiver.envoy.als.AbstractALSAnalyzer; @@ -79,9 +80,9 @@ public class K8sALSServiceMeshHTTPAnalysis extends AbstractALSAnalyzer { return result; } return analyzeSideCar(result, entry); + case WAYPOINT: + return analyzeWaypoint(result, identifier, entry); case NONE: - // For the Ambient Istio with waypoint mode, the role is NONE. - // The analysis should keep the result from other roles. return result; } @@ -163,16 +164,68 @@ public class K8sALSServiceMeshHTTPAnalysis extends AbstractALSAnalyzer { return previousResult; } - final SocketAddress downstreamRemoteAddressSocketAddress = downstreamRemoteAddress.getSocketAddress(); - final SocketAddress downstreamLocalAddressSocketAddress = downstreamLocalAddress.getSocketAddress(); - final ServiceMetaInfo ingress = find(downstreamLocalAddressSocketAddress.getAddress()); + return buildUpstreamDownstreamMetrics(previousResult, entry, + Addresses.getAddressIP(downstreamRemoteAddress), + Addresses.getAddressIP(downstreamLocalAddress), + Addresses.getAddressIP(upstreamRemoteAddress), NON_TLS); + } + + protected Result analyzeWaypoint(Result previousResult, StreamAccessLogsMessage.Identifier identifier, final HTTPAccessLogEntry entry) { + if (!entry.hasCommonProperties()) { + return previousResult; + } + if (previousResult.hasUpstreamMetrics() && previousResult.hasDownstreamMetrics()) { + return previousResult; + } + + final String waypointIP = getWaypointIP(identifier); + final AccessLogCommon properties = entry.getCommonProperties(); + final Address downstreamRemoteAddress = properties.hasDownstreamDirectRemoteAddress() && + Addresses.isValid(properties.getDownstreamDirectRemoteAddress()) ? + properties.getDownstreamDirectRemoteAddress() : properties.getDownstreamRemoteAddress(); + final Address upstreamRemoteAddress = properties.getUpstreamRemoteAddress(); + if (!isValid(downstreamRemoteAddress) || !isValid(upstreamRemoteAddress)) { + return previousResult; + } + + return buildUpstreamDownstreamMetrics(previousResult, entry, + Addresses.getAddressIP(downstreamRemoteAddress), + waypointIP, + Addresses.getAddressIP(upstreamRemoteAddress), null); + } + + protected String getWaypointIP(StreamAccessLogsMessage.Identifier identifier) { + final Value instanceIps = identifier.getNode().getMetadata().getFieldsMap().get("INSTANCE_IPS"); + if (instanceIps != null && instanceIps.hasStringValue()) { + final String[] split = instanceIps.getStringValue().split(":", 2); + if (split.length == 2) { + return split[0]; + } + } + + final String nodeId = identifier.getNode().getId(); + final String[] nodeInfo = nodeId.split("~", 3); + if (nodeInfo.length != 3) { + return null; + } + return nodeInfo[1]; + } + + protected Result buildUpstreamDownstreamMetrics(Result previousResult, final HTTPAccessLogEntry entry, + String downStreamRemoteAddr, String downStreamLocalAddr, + String upstreamRemoteAddr, String upstreamMetricsTLS) { + if (StringUtil.isEmpty(downStreamRemoteAddr) || StringUtil.isEmpty(downStreamLocalAddr) || StringUtil.isEmpty(upstreamRemoteAddr)) { + return previousResult; + } + + final ServiceMetaInfo ingress = find(downStreamLocalAddr); final var newResult = previousResult.toBuilder(); final var previousMetrics = previousResult.getMetrics(); final var previousHttpMetrics = previousMetrics.getHttpMetricsBuilder(); if (!previousResult.hasDownstreamMetrics()) { - final ServiceMetaInfo outside = find(downstreamRemoteAddressSocketAddress.getAddress()); + final ServiceMetaInfo outside = find(downStreamRemoteAddr); final HTTPServiceMeshMetric.Builder metric = newAdapter(entry, outside, ingress).adaptToDownstreamMetrics(); @@ -182,14 +235,15 @@ public class K8sALSServiceMeshHTTPAnalysis extends AbstractALSAnalyzer { } if (!previousResult.hasUpstreamMetrics()) { - final SocketAddress upstreamRemoteAddressSocketAddress = upstreamRemoteAddress.getSocketAddress(); - final ServiceMetaInfo targetService = find(upstreamRemoteAddressSocketAddress.getAddress()); + final ServiceMetaInfo targetService = find(upstreamRemoteAddr); - final HTTPServiceMeshMetric.Builder outboundMetric = + HTTPServiceMeshMetric.Builder outboundMetric = newAdapter(entry, ingress, targetService) - .adaptToUpstreamMetrics() - // Can't parse it from tls properties, leave it to Server side. - .setTlsMode(NON_TLS); + .adaptToUpstreamMetrics(); + if (StringUtil.isNotEmpty(upstreamMetricsTLS)) { + // Can't parse it from tls properties, leave it to Server side. + outboundMetric = outboundMetric.setTlsMode(NON_TLS); + } log.debug("Transformed ingress outbound mesh metric {}", outboundMetric); previousHttpMetrics.addMetrics(outboundMetric); diff --git a/oap-server/server-receiver-plugin/receiver-proto/src/main/proto/envoy/config/core/v3/address.proto b/oap-server/server-receiver-plugin/receiver-proto/src/main/proto/envoy/config/core/v3/address.proto index 8228450eb9..cbe57ff16e 100644 --- a/oap-server/server-receiver-plugin/receiver-proto/src/main/proto/envoy/config/core/v3/address.proto +++ b/oap-server/server-receiver-plugin/receiver-proto/src/main/proto/envoy/config/core/v3/address.proto @@ -13,6 +13,7 @@ import "validate/validate.proto"; option java_package = "io.envoyproxy.envoy.config.core.v3"; option java_outer_classname = "AddressProto"; option java_multiple_files = true; +option go_package = "github.com/envoyproxy/go-control-plane/envoy/config/core/v3;corev3"; option (udpa.annotations.file_status).package_version_status = ACTIVE; // [#protodoc-title: Network addresses] @@ -30,19 +31,24 @@ message Pipe { uint32 mode = 2 [(validate.rules).uint32 = {lte: 511}]; } -// [#not-implemented-hide:] The address represents an envoy internal listener. -// TODO(lambdai): Make this address available for listener and endpoint. -// TODO(asraa): When address available, remove workaround from test/server/server_fuzz_test.cc:30. +// The address represents an envoy internal listener. +// [#comment: TODO(asraa): When address available, remove workaround from test/server/server_fuzz_test.cc:30.] message EnvoyInternalAddress { oneof address_name_specifier { option (validate.required) = true; - // [#not-implemented-hide:] The :ref:`listener name ` of the destination internal listener. + // Specifies the :ref:`name ` of the + // internal listener. string server_listener_name = 1; } + + // Specifies an endpoint identifier to distinguish between multiple endpoints for the same internal listener in a + // single upstream pool. Only used in the upstream addresses for tracking changes to individual endpoints. This, for + // example, may be set to the final destination IP for the target internal listener. + string endpoint_id = 2; } -// [#next-free-field: 7] +// [#next-free-field: 8] message SocketAddress { option (udpa.annotations.versioning).previous_message_type = "envoy.api.v2.core.SocketAddress"; @@ -57,13 +63,13 @@ message SocketAddress { // to the address. An empty address is not allowed. Specify ``0.0.0.0`` or ``::`` // to bind to any address. [#comment:TODO(zuercher) reinstate when implemented: // It is possible to distinguish a Listener address via the prefix/suffix matching - // in :ref:`FilterChainMatch `.] When used - // within an upstream :ref:`BindConfig `, the address + // in :ref:`FilterChainMatch `.] When used + // within an upstream :ref:`BindConfig `, the address // controls the source address of outbound connections. For :ref:`clusters - // `, the cluster type determines whether the - // address must be an IP (*STATIC* or *EDS* clusters) or a hostname resolved by DNS - // (*STRICT_DNS* or *LOGICAL_DNS* clusters). Address resolution can be customized - // via :ref:`resolver_name `. + // `, the cluster type determines whether the + // address must be an IP (``STATIC`` or ``EDS`` clusters) or a hostname resolved by DNS + // (``STRICT_DNS`` or ``LOGICAL_DNS`` clusters). Address resolution can be customized + // via :ref:`resolver_name `. string address = 2 [(validate.rules).string = {min_len: 1}]; oneof port_specifier { @@ -72,7 +78,7 @@ message SocketAddress { uint32 port_value = 3 [(validate.rules).uint32 = {lte: 65535}]; // This is only valid if :ref:`resolver_name - // ` is specified below and the + // ` is specified below and the // named resolver is capable of named port resolution. string named_port = 4; } @@ -81,7 +87,7 @@ message SocketAddress { // this is empty, a context dependent default applies. If the address is a concrete // IP address, no resolution will occur. If address is a hostname this // should be set for resolution other than DNS. Specifying a custom resolver with - // *STRICT_DNS* or *LOGICAL_DNS* will generate an error at runtime. + // ``STRICT_DNS`` or ``LOGICAL_DNS`` will generate an error at runtime. string resolver_name = 5; // When binding to an IPv6 address above, this enables `IPv4 compatibility @@ -89,6 +95,11 @@ message SocketAddress { // allow both IPv4 and IPv6 connections, with peer IPv4 addresses mapped into // IPv6 space as ``::FFFF:``. bool ipv4_compat = 6; + + // The Linux network namespace to bind the socket to. If this is set, Envoy will + // create the socket in the specified network namespace. Only supported on Linux. + // [#not-implemented-hide:] + string network_namespace_filepath = 7; } message TcpKeepalive { @@ -109,17 +120,18 @@ message TcpKeepalive { google.protobuf.UInt32Value keepalive_interval = 3; } +// [#next-free-field: 7] message BindConfig { option (udpa.annotations.versioning).previous_message_type = "envoy.api.v2.core.BindConfig"; // The address to bind to when creating a socket. - SocketAddress source_address = 1 [(validate.rules).message = {required: true}]; + SocketAddress source_address = 1; - // Whether to set the *IP_FREEBIND* option when creating the socket. When this + // Whether to set the ``IP_FREEBIND`` option when creating the socket. When this // flag is set to true, allows the :ref:`source_address - // ` to be an IP address + // ` to be an IP address // that is not configured on the system running Envoy. When this flag is set - // to false, the option *IP_FREEBIND* is disabled on the socket. When this + // to false, the option ``IP_FREEBIND`` is disabled on the socket. When this // flag is not set (default), the socket is not modified, i.e. the option is // neither enabled nor disabled. google.protobuf.BoolValue freebind = 2; @@ -142,7 +154,8 @@ message Address { Pipe pipe = 2; - // [#not-implemented-hide:] + // Specifies a user-space address handled by :ref:`internal listeners + // `. EnvoyInternalAddress envoy_internal_address = 3; } } @@ -155,6 +168,6 @@ message CidrRange { // IPv4 or IPv6 address, e.g. ``192.0.0.0`` or ``2001:db8::``. string address_prefix = 1 [(validate.rules).string = {min_len: 1}]; - // Length of prefix, e.g. 0, 32. + // Length of prefix, e.g. 0, 32. Defaults to 0 when unset. google.protobuf.UInt32Value prefix_len = 2 [(validate.rules).uint32 = {lte: 128}]; }