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

Author: mrproliu

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
 

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 @@ default Role identify(StreamAccessLogsMessage.Identifier alsIdentifier, Role def
             return Role.PROXY;
         } else if (id.startsWith("sidecar~")) {
             return Role.SIDECAR;
+        } else if (id.startsWith("waypoint~")) {
+            return Role.WAYPOINT;
         }
         return defaultRole;
     }

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,
 }

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

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 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 Result analysis(
                     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 @@ protected Result analyzeProxy(Result previousResult, final HTTPAccessLogEntry en
             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 @@ protected Result analyzeProxy(Result previousResult, final HTTPAccessLogEntry en
         }
 
         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);

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 <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,14 +87,19 @@ 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
   // <https://tools.ietf.org/html/rfc3493#page-11>`_. Binding to ``::`` will
   // 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}];
 }