kazusa
/
skywalking
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Support analysis waypoint metrics in Envoy ALS receiver (#13244)

This commit is contained in:
mrproliu 2025-05-13 17:01:05 +08:00 committed by GitHub
parent 3739add599
commit a782fc106a
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 132 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
docs/en/changes/changes.md
View File

@ -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

2
oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/AccessLogAnalyzer.java
View File

@ -65,6 +65,8 @@ public interface AccessLogAnalyzer<E> {
return Role.PROXY;
} else if (id.startsWith("sidecar~")) {
return Role.SIDECAR;
} else if (id.startsWith("waypoint~")) {
return Role.WAYPOINT;
}
return defaultRole;
}

6
oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/Role.java
View File

@ -33,5 +33,9 @@ public enum Role {
/**
* Sidecar in mesh
*/
SIDECAR
SIDECAR,
/**
* Waypoint in ambient mesh
*/
WAYPOINT,
}

29
oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/k8s/Addresses.java
View File

@ -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;
}
}

80
oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/k8s/K8sALSServiceMeshHTTPAnalysis.java
View File

@ -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);

51
oap-server/server-receiver-plugin/receiver-proto/src/main/proto/envoy/config/core/v3/address.proto
View File

@ -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 <envoy_api_field_config.listener.v3.Listener.name>` of the destination internal listener.
// Specifies the :ref:`name <envoy_v3_api_field_config.listener.v3.Listener.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 <envoy_api_msg_config.listener.v3.FilterChainMatch>`.] When used
// within an upstream :ref:`BindConfig <envoy_api_msg_config.core.v3.BindConfig>`, the address
// in :ref:`FilterChainMatch <envoy_v3_api_msg_config.listener.v3.FilterChainMatch>`.] When used
// within an upstream :ref:`BindConfig <envoy_v3_api_msg_config.core.v3.BindConfig>`, the address
// controls the source address of outbound connections. For :ref:`clusters
// <envoy_api_msg_config.cluster.v3.Cluster>`, 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 <envoy_api_field_config.core.v3.SocketAddress.resolver_name>`.
// <envoy_v3_api_msg_config.cluster.v3.Cluster>`, 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 <envoy_v3_api_field_config.core.v3.SocketAddress.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
// <envoy_api_field_config.core.v3.SocketAddress.resolver_name>` is specified below and the
// <envoy_v3_api_field_config.core.v3.SocketAddress.resolver_name>` 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:<IPv4-address>``.
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
// <envoy_api_field_config.cluster.v3.UpstreamBindConfig.source_address>` to be an IP address
// <envoy_v3_api_field_config.core.v3.BindConfig.source_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
// <envoy_v3_api_field_config.listener.v3.Listener.internal_listener>`.
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}];
}