From 16f940b5ce411d1ccd8d50781b6d519c7b5feaa6 Mon Sep 17 00:00:00 2001 From: kezhenxu94 Date: Sun, 27 Aug 2023 12:20:35 +0800 Subject: [PATCH] Check results in ALS as per downstream/upstream instead of per log (#11265) --- docs/en/changes/changes.md | 1 + .../receiver/envoy/als/AccessLogAnalyzer.java | 22 +++-- .../k8s/K8sALSServiceMeshHTTPAnalysis.java | 82 +++++++++++-------- .../als/mx/MetaExchangeALSHTTPAnalyzer.java | 36 ++++---- .../tcp/k8s/K8sALSServiceMeshTCPAnalysis.java | 65 +++++++++------ .../mx/MetaExchangeTCPAccessLogAnalyzer.java | 35 +++++--- 6 files changed, 148 insertions(+), 93 deletions(-) diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md index ed3be1e5ca..9ce6c186af 100644 --- a/docs/en/changes/changes.md +++ b/docs/en/changes/changes.md @@ -76,6 +76,7 @@ * Enhance the `serviceRelation` in MAL by adding settings for the `delimiter` and `component` fields. * [Breaking change] Support MQE in the Alerting. The Alarm Rules configuration(alarm-settings.yml), add `expression` field and remove `metrics-name/count/threshold/op/only-as-condition` fields and remove `composite-rules` configuration. +* Check results in ALS as per downstream/upstream instead of per log. #### 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 ac9327bdb7..187f023213 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 @@ -18,6 +18,8 @@ package org.apache.skywalking.oap.server.receiver.envoy.als; +import lombok.Getter; +import lombok.experimental.Accessors; 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.module.ModuleStartException; @@ -34,7 +36,7 @@ public interface AccessLogAnalyzer { /** * The method works as a chain of analyzers. Logs are processed sequentially by analyzers one by one, the results of the previous analyzer are passed into the current one. - * + *

* To do fast-success, the analyzer could simply check the results of the previous analyzer and return if not empty. * * @param result of the previous analyzer. @@ -68,8 +70,8 @@ public interface AccessLogAnalyzer { } @Data - @Builder - static class Result { + @Builder(toBuilder = true) + class Result { /** * The service representing the Envoy node. */ @@ -78,10 +80,16 @@ public interface AccessLogAnalyzer { /** * The analyzed metrics result. */ - private ServiceMeshMetrics.Builder metrics; + @Builder.Default + private ServiceMeshMetrics.Builder metrics = ServiceMeshMetrics.newBuilder(); - public boolean hasResult() { - return metrics != null && (metrics.getHttpMetrics().getMetricsCount() > 0 || metrics.getTcpMetrics().getMetricsCount() > 0); - } + @Accessors(fluent = true) + private boolean hasDownstreamMetrics; + @Accessors(fluent = true) + private boolean hasUpstreamMetrics; + + @Getter(lazy = true) + @Accessors(fluent = true) + private final boolean hasResult = hasDownstreamMetrics || hasUpstreamMetrics; } } 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 01cba845eb..2c7e24c265 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 @@ -26,7 +26,6 @@ import io.envoyproxy.envoy.service.accesslog.v3.StreamAccessLogsMessage; import lombok.SneakyThrows; import lombok.extern.slf4j.Slf4j; import org.apache.skywalking.apm.network.servicemesh.v3.HTTPServiceMeshMetric; -import org.apache.skywalking.apm.network.servicemesh.v3.HTTPServiceMeshMetrics; import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetrics; import org.apache.skywalking.oap.server.library.module.ModuleManager; import org.apache.skywalking.oap.server.receiver.envoy.EnvoyMetricReceiverConfig; @@ -72,31 +71,29 @@ public class K8sALSServiceMeshHTTPAnalysis extends AbstractALSAnalyzer { final HTTPAccessLogEntry entry, final Role role ) { - if (result.hasResult()) { - return result; - } switch (role) { case PROXY: - return analyzeProxy(entry); + return analyzeProxy(result, entry); case SIDECAR: - return analyzeSideCar(entry); + if (result.hasResult()) { + return result; + } + return analyzeSideCar(result, entry); } return Result.builder().build(); } - protected Result analyzeSideCar(final HTTPAccessLogEntry entry) { + protected Result analyzeSideCar(final Result previousResult, final HTTPAccessLogEntry entry) { if (!entry.hasCommonProperties()) { - return Result.builder().build(); + return previousResult; } final AccessLogCommon properties = entry.getCommonProperties(); final String cluster = properties.getUpstreamCluster(); if (isBlank(cluster)) { - return Result.builder().build(); + return previousResult; } - final HTTPServiceMeshMetrics.Builder sources = HTTPServiceMeshMetrics.newBuilder(); - final Address downstreamRemoteAddress = properties.hasDownstreamDirectRemoteAddress() ? properties.getDownstreamDirectRemoteAddress() @@ -104,10 +101,13 @@ public class K8sALSServiceMeshHTTPAnalysis extends AbstractALSAnalyzer { final ServiceMetaInfo downstreamService = find(downstreamRemoteAddress.getSocketAddress().getAddress()); final Address downstreamLocalAddress = properties.getDownstreamLocalAddress(); if (!isValid(downstreamRemoteAddress) || !isValid(downstreamLocalAddress)) { - return Result.builder().build(); + return previousResult; } final ServiceMetaInfo localService = find(downstreamLocalAddress.getSocketAddress().getAddress()); + final var result = Result.builder(); + final var previousMetrics = previousResult.getMetrics(); + final var sources = previousMetrics.getHttpMetricsBuilder(); if (cluster.startsWith("inbound|")) { // Server side final HTTPServiceMeshMetric.Builder metrics; @@ -120,15 +120,15 @@ public class K8sALSServiceMeshHTTPAnalysis extends AbstractALSAnalyzer { } else { // sidecar -> sidecar(server side) metrics = newAdapter(entry, downstreamService, localService).adaptToDownstreamMetrics(); - log.debug("Transformed sidecar->sidecar(server side) inbound mesh metrics {}", metrics); } sources.addMetrics(metrics); + result.hasDownstreamMetrics(true); } else if (cluster.startsWith("outbound|")) { // sidecar(client side) -> sidecar final Address upstreamRemoteAddress = properties.getUpstreamRemoteAddress(); if (!isValid(upstreamRemoteAddress)) { - return Result.builder().metrics(ServiceMeshMetrics.newBuilder().setHttpMetrics(sources)).service(localService).build(); + return result.metrics(ServiceMeshMetrics.newBuilder().setHttpMetrics(sources)).service(localService).build(); } final ServiceMetaInfo destService = find(upstreamRemoteAddress.getSocketAddress().getAddress()); @@ -136,49 +136,63 @@ public class K8sALSServiceMeshHTTPAnalysis extends AbstractALSAnalyzer { log.debug("Transformed sidecar->sidecar(server side) inbound mesh metric {}", metric); sources.addMetrics(metric); + result.hasUpstreamMetrics(true); } - return Result.builder().metrics(ServiceMeshMetrics.newBuilder().setHttpMetrics(sources)).service(localService).build(); + return result.metrics(previousMetrics.setHttpMetrics(sources)).service(localService).build(); } - protected Result analyzeProxy(final HTTPAccessLogEntry entry) { + protected Result analyzeProxy(Result previousResult, final HTTPAccessLogEntry entry) { if (!entry.hasCommonProperties()) { - return Result.builder().build(); + return previousResult; } + if (previousResult.hasUpstreamMetrics() && previousResult.hasDownstreamMetrics()) { + return previousResult; + } + final AccessLogCommon properties = entry.getCommonProperties(); final Address downstreamLocalAddress = properties.getDownstreamLocalAddress(); final Address downstreamRemoteAddress = properties.hasDownstreamDirectRemoteAddress() ? properties.getDownstreamDirectRemoteAddress() : properties.getDownstreamRemoteAddress(); final Address upstreamRemoteAddress = properties.getUpstreamRemoteAddress(); if (!isValid(downstreamLocalAddress) || !isValid(downstreamRemoteAddress) || !isValid(upstreamRemoteAddress)) { - return Result.builder().build(); + return previousResult; } - final HTTPServiceMeshMetrics.Builder result = HTTPServiceMeshMetrics.newBuilder(); final SocketAddress downstreamRemoteAddressSocketAddress = downstreamRemoteAddress.getSocketAddress(); - final ServiceMetaInfo outside = find(downstreamRemoteAddressSocketAddress.getAddress()); - final SocketAddress downstreamLocalAddressSocketAddress = downstreamLocalAddress.getSocketAddress(); final ServiceMetaInfo ingress = find(downstreamLocalAddressSocketAddress.getAddress()); - final HTTPServiceMeshMetric.Builder metric = newAdapter(entry, outside, ingress).adaptToDownstreamMetrics(); + final var newResult = previousResult.toBuilder(); + final var previousMetrics = previousResult.getMetrics(); + final var previousHttpMetrics = previousMetrics.getHttpMetricsBuilder(); - log.debug("Transformed ingress inbound mesh metric {}", metric); - result.addMetrics(metric); + if (!previousResult.hasDownstreamMetrics()) { + final ServiceMetaInfo outside = find(downstreamRemoteAddressSocketAddress.getAddress()); - final SocketAddress upstreamRemoteAddressSocketAddress = upstreamRemoteAddress.getSocketAddress(); - final ServiceMetaInfo targetService = find(upstreamRemoteAddressSocketAddress.getAddress()); + final HTTPServiceMeshMetric.Builder metric = newAdapter(entry, outside, ingress).adaptToDownstreamMetrics(); - final HTTPServiceMeshMetric.Builder outboundMetric = - newAdapter(entry, ingress, targetService) - .adaptToUpstreamMetrics() - // Can't parse it from tls properties, leave it to Server side. - .setTlsMode(NON_TLS); + log.debug("Transformed ingress inbound mesh metric {}", metric); + previousHttpMetrics.addMetrics(metric); + newResult.hasDownstreamMetrics(true); + } - log.debug("Transformed ingress outbound mesh metric {}", outboundMetric); - result.addMetrics(outboundMetric); + if (!previousResult.hasUpstreamMetrics()) { + final SocketAddress upstreamRemoteAddressSocketAddress = upstreamRemoteAddress.getSocketAddress(); + final ServiceMetaInfo targetService = find(upstreamRemoteAddressSocketAddress.getAddress()); - return Result.builder().metrics(ServiceMeshMetrics.newBuilder().setHttpMetrics(result)).service(ingress).build(); + final HTTPServiceMeshMetric.Builder outboundMetric = + newAdapter(entry, ingress, targetService) + .adaptToUpstreamMetrics() + // Can't parse it from tls properties, leave it to Server side. + .setTlsMode(NON_TLS); + + log.debug("Transformed ingress outbound mesh metric {}", outboundMetric); + previousHttpMetrics.addMetrics(outboundMetric); + newResult.hasUpstreamMetrics(true); + } + + return newResult.metrics(previousMetrics.setHttpMetrics(previousHttpMetrics)).service(ingress).build(); } /** diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/mx/MetaExchangeALSHTTPAnalyzer.java b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/mx/MetaExchangeALSHTTPAnalyzer.java index 10410f3f3d..145e003b02 100644 --- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/mx/MetaExchangeALSHTTPAnalyzer.java +++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/mx/MetaExchangeALSHTTPAnalyzer.java @@ -20,16 +20,10 @@ package org.apache.skywalking.oap.server.receiver.envoy.als.mx; import com.google.protobuf.Any; import com.google.protobuf.TextFormat; -import io.envoyproxy.envoy.data.accesslog.v3.AccessLogCommon; import io.envoyproxy.envoy.data.accesslog.v3.HTTPAccessLogEntry; import io.envoyproxy.envoy.service.accesslog.v3.StreamAccessLogsMessage; -import java.util.Base64; -import java.util.Map; -import java.util.concurrent.atomic.AtomicBoolean; import lombok.extern.slf4j.Slf4j; import org.apache.skywalking.apm.network.servicemesh.v3.HTTPServiceMeshMetric; -import org.apache.skywalking.apm.network.servicemesh.v3.HTTPServiceMeshMetrics; -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.module.ModuleStartException; import org.apache.skywalking.oap.server.receiver.envoy.EnvoyMetricReceiverConfig; @@ -37,6 +31,9 @@ import org.apache.skywalking.oap.server.receiver.envoy.als.AbstractALSAnalyzer; import org.apache.skywalking.oap.server.receiver.envoy.als.Role; import org.apache.skywalking.oap.server.receiver.envoy.als.ServiceMetaInfo; +import java.util.Base64; +import java.util.concurrent.atomic.AtomicBoolean; + import static org.apache.skywalking.oap.server.core.Const.TLS_MODE.NON_TLS; @Slf4j @@ -72,7 +69,7 @@ public class MetaExchangeALSHTTPAnalyzer extends AbstractALSAnalyzer { final HTTPAccessLogEntry entry, final Role role ) { - if (previousResult.hasResult()) { + if (previousResult.hasUpstreamMetrics() && previousResult.hasDownstreamMetrics()) { return previousResult; } if (!entry.hasCommonProperties()) { @@ -85,14 +82,16 @@ public class MetaExchangeALSHTTPAnalyzer extends AbstractALSAnalyzer { log.error("Failed to inflate the ServiceMetaInfo from identifier.node.metadata. ", e); return previousResult; } - final AccessLogCommon properties = entry.getCommonProperties(); - final Map stateMap = properties.getFilterStateObjectsMap(); + final var properties = entry.getCommonProperties(); + final var stateMap = properties.getFilterStateObjectsMap(); + final var result = previousResult.toBuilder(); if (stateMap.isEmpty()) { - return Result.builder().service(currSvc).build(); + return result.service(currSvc).build(); } - final HTTPServiceMeshMetrics.Builder httpMetrics = HTTPServiceMeshMetrics.newBuilder(); - final AtomicBoolean downstreamExists = new AtomicBoolean(); + final var previousMetrics = previousResult.getMetrics(); + final var httpMetrics = previousMetrics.getHttpMetricsBuilder(); + final var downstreamExists = new AtomicBoolean(); stateMap.forEach((key, value) -> { if (!key.equals(UPSTREAM_KEY) && !key.equals(DOWNSTREAM_KEY)) { return; @@ -107,30 +106,39 @@ public class MetaExchangeALSHTTPAnalyzer extends AbstractALSAnalyzer { final HTTPServiceMeshMetric.Builder metrics; switch (key) { case UPSTREAM_KEY: + if (previousResult.hasUpstreamMetrics()) { + break; + } metrics = newAdapter(entry, currSvc, svc).adaptToUpstreamMetrics().setTlsMode(NON_TLS); if (log.isDebugEnabled()) { log.debug("Transformed a {} outbound mesh metrics {}", role, TextFormat.shortDebugString(metrics)); } httpMetrics.addMetrics(metrics); + result.hasUpstreamMetrics(true); break; case DOWNSTREAM_KEY: + if (previousResult.hasDownstreamMetrics()) { + break; + } metrics = newAdapter(entry, svc, currSvc).adaptToDownstreamMetrics(); if (log.isDebugEnabled()) { log.debug("Transformed a {} inbound mesh metrics {}", role, TextFormat.shortDebugString(metrics)); } httpMetrics.addMetrics(metrics); downstreamExists.set(true); + result.hasDownstreamMetrics(true); break; } }); if (role.equals(Role.PROXY) && !downstreamExists.get()) { - final HTTPServiceMeshMetric.Builder metric = newAdapter(entry, config.serviceMetaInfoFactory().unknown(), currSvc).adaptToDownstreamMetrics(); + final var metric = newAdapter(entry, config.serviceMetaInfoFactory().unknown(), currSvc).adaptToDownstreamMetrics(); if (log.isDebugEnabled()) { log.debug("Transformed a {} inbound mesh metric {}", role, TextFormat.shortDebugString(metric)); } httpMetrics.addMetrics(metric); + result.hasDownstreamMetrics(true); } - return Result.builder().metrics(ServiceMeshMetrics.newBuilder().setHttpMetrics(httpMetrics)).service(currSvc).build(); + return result.metrics(previousMetrics.setHttpMetrics(httpMetrics)).service(currSvc).build(); } protected ServiceMetaInfo adaptToServiceMetaInfo(final Any value) throws Exception { diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/tcp/k8s/K8sALSServiceMeshTCPAnalysis.java b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/tcp/k8s/K8sALSServiceMeshTCPAnalysis.java index 5ce1d1dd33..3aeee112e2 100644 --- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/tcp/k8s/K8sALSServiceMeshTCPAnalysis.java +++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/tcp/k8s/K8sALSServiceMeshTCPAnalysis.java @@ -26,9 +26,7 @@ import io.envoyproxy.envoy.data.accesslog.v3.TCPAccessLogEntry; import io.envoyproxy.envoy.service.accesslog.v3.StreamAccessLogsMessage; import lombok.SneakyThrows; import lombok.extern.slf4j.Slf4j; -import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetrics; import org.apache.skywalking.apm.network.servicemesh.v3.TCPServiceMeshMetric; -import org.apache.skywalking.apm.network.servicemesh.v3.TCPServiceMeshMetrics; import org.apache.skywalking.oap.server.library.module.ModuleManager; import org.apache.skywalking.oap.server.receiver.envoy.EnvoyMetricReceiverConfig; import org.apache.skywalking.oap.server.receiver.envoy.ServiceMetaInfoFactory; @@ -72,13 +70,13 @@ public class K8sALSServiceMeshTCPAnalysis extends AbstractTCPAccessLogAnalyzer { final TCPAccessLogEntry entry, final Role role ) { - if (previousResult.getMetrics() != null && previousResult.getMetrics().hasTcpMetrics()) { - return previousResult; - } switch (role) { case PROXY: return analyzeProxy(previousResult, entry); case SIDECAR: + if (previousResult.hasResult()) { + return previousResult; + } return analyzeSideCar(previousResult, entry); } @@ -95,7 +93,9 @@ public class K8sALSServiceMeshTCPAnalysis extends AbstractTCPAccessLogAnalyzer { return previousResult; } - final TCPServiceMeshMetrics.Builder sources = TCPServiceMeshMetrics.newBuilder(); + final var newResult = previousResult.toBuilder(); + final var previousMetrics = previousResult.getMetrics(); + final var sources = previousMetrics.getTcpMetricsBuilder(); final Address downstreamRemoteAddress = properties.hasDownstreamDirectRemoteAddress() @@ -121,6 +121,7 @@ public class K8sALSServiceMeshTCPAnalysis extends AbstractTCPAccessLogAnalyzer { log.debug("Transformed sidecar->sidecar(server side) inbound mesh metrics {}", metrics); } sources.addMetrics(metrics); + newResult.hasDownstreamMetrics(true); } else if (cluster.startsWith("outbound|")) { // sidecar(client side) -> sidecar final Address upstreamRemoteAddress = properties.getUpstreamRemoteAddress(); @@ -130,15 +131,20 @@ public class K8sALSServiceMeshTCPAnalysis extends AbstractTCPAccessLogAnalyzer { log.debug("Transformed sidecar->sidecar(server side) inbound mesh metric {}", metric); sources.addMetrics(metric); + newResult.hasUpstreamMetrics(true); } - return Result.builder().metrics(ServiceMeshMetrics.newBuilder().setTcpMetrics(sources)).service(localService).build(); + return newResult.metrics(previousMetrics.setTcpMetrics(sources)).service(localService).build(); } protected Result analyzeProxy(final Result previousResult, final TCPAccessLogEntry entry) { if (!entry.hasCommonProperties()) { return previousResult; } + if (previousResult.hasUpstreamMetrics() && previousResult.hasDownstreamMetrics()) { + return previousResult; + } + final AccessLogCommon properties = entry.getCommonProperties(); final Address downstreamLocalAddress = properties.getDownstreamLocalAddress(); final Address downstreamRemoteAddress = properties.hasDownstreamDirectRemoteAddress() ? @@ -148,31 +154,40 @@ public class K8sALSServiceMeshTCPAnalysis extends AbstractTCPAccessLogAnalyzer { return previousResult; } - final TCPServiceMeshMetrics.Builder metrics = TCPServiceMeshMetrics.newBuilder(); - final SocketAddress downstreamRemoteAddressSocketAddress = downstreamRemoteAddress.getSocketAddress(); - final ServiceMetaInfo outside = find(downstreamRemoteAddressSocketAddress.getAddress()); + final var downstreamLocalAddressSocketAddress = downstreamLocalAddress.getSocketAddress(); + final var ingress = find(downstreamLocalAddressSocketAddress.getAddress()); - final SocketAddress downstreamLocalAddressSocketAddress = downstreamLocalAddress.getSocketAddress(); - final ServiceMetaInfo ingress = find(downstreamLocalAddressSocketAddress.getAddress()); + final var newResult = previousResult.toBuilder(); + final var previousMetrics = previousResult.getMetrics(); + final var metrics = previousMetrics.getTcpMetricsBuilder(); - final TCPServiceMeshMetric.Builder metric = newAdapter(entry, outside, ingress).adaptToDownstreamMetrics(); + if (!previousResult.hasDownstreamMetrics()) { + final SocketAddress downstreamRemoteAddressSocketAddress = downstreamRemoteAddress.getSocketAddress(); + final ServiceMetaInfo outside = find(downstreamRemoteAddressSocketAddress.getAddress()); - log.debug("Transformed ingress inbound mesh metric {}", metric); - metrics.addMetrics(metric); + final TCPServiceMeshMetric.Builder metric = newAdapter(entry, outside, ingress).adaptToDownstreamMetrics(); - final SocketAddress upstreamRemoteAddressSocketAddress = upstreamRemoteAddress.getSocketAddress(); - final ServiceMetaInfo targetService = find(upstreamRemoteAddressSocketAddress.getAddress()); + log.debug("Transformed ingress inbound mesh metric {}", metric); + metrics.addMetrics(metric); + newResult.hasDownstreamMetrics(true); + } - final TCPServiceMeshMetric.Builder outboundMetric = - newAdapter(entry, ingress, targetService) - .adaptToUpstreamMetrics() - // Can't parse it from tls properties, leave it to Server side. - .setTlsMode(NON_TLS); + if (!previousResult.hasUpstreamMetrics()) { + final SocketAddress upstreamRemoteAddressSocketAddress = upstreamRemoteAddress.getSocketAddress(); + final ServiceMetaInfo targetService = find(upstreamRemoteAddressSocketAddress.getAddress()); - log.debug("Transformed ingress outbound mesh metric {}", outboundMetric); - metrics.addMetrics(outboundMetric); + final TCPServiceMeshMetric.Builder outboundMetric = + newAdapter(entry, ingress, targetService) + .adaptToUpstreamMetrics() + // Can't parse it from tls properties, leave it to Server side. + .setTlsMode(NON_TLS); - return Result.builder().metrics(ServiceMeshMetrics.newBuilder().setTcpMetrics(metrics)).service(ingress).build(); + log.debug("Transformed ingress outbound mesh metric {}", outboundMetric); + metrics.addMetrics(outboundMetric); + newResult.hasUpstreamMetrics(true); + } + + return newResult.metrics(previousMetrics.setTcpMetrics(metrics)).service(ingress).build(); } /** diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/tcp/mx/MetaExchangeTCPAccessLogAnalyzer.java b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/tcp/mx/MetaExchangeTCPAccessLogAnalyzer.java index e1a6cd6c7b..7f53439739 100644 --- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/tcp/mx/MetaExchangeTCPAccessLogAnalyzer.java +++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/tcp/mx/MetaExchangeTCPAccessLogAnalyzer.java @@ -23,13 +23,8 @@ import com.google.protobuf.TextFormat; import io.envoyproxy.envoy.data.accesslog.v3.AccessLogCommon; import io.envoyproxy.envoy.data.accesslog.v3.TCPAccessLogEntry; import io.envoyproxy.envoy.service.accesslog.v3.StreamAccessLogsMessage; -import java.util.Base64; -import java.util.Map; -import java.util.concurrent.atomic.AtomicBoolean; import lombok.extern.slf4j.Slf4j; -import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetrics; import org.apache.skywalking.apm.network.servicemesh.v3.TCPServiceMeshMetric; -import org.apache.skywalking.apm.network.servicemesh.v3.TCPServiceMeshMetrics; import org.apache.skywalking.oap.server.library.module.ModuleManager; import org.apache.skywalking.oap.server.library.module.ModuleStartException; import org.apache.skywalking.oap.server.receiver.envoy.EnvoyMetricReceiverConfig; @@ -39,6 +34,10 @@ import org.apache.skywalking.oap.server.receiver.envoy.als.mx.FieldsHelper; import org.apache.skywalking.oap.server.receiver.envoy.als.mx.ServiceMetaInfoAdapter; import org.apache.skywalking.oap.server.receiver.envoy.als.tcp.AbstractTCPAccessLogAnalyzer; +import java.util.Base64; +import java.util.Map; +import java.util.concurrent.atomic.AtomicBoolean; + import static org.apache.skywalking.oap.server.core.Const.TLS_MODE.NON_TLS; import static org.apache.skywalking.oap.server.receiver.envoy.als.mx.MetaExchangeALSHTTPAnalyzer.DOWNSTREAM_KEY; import static org.apache.skywalking.oap.server.receiver.envoy.als.mx.MetaExchangeALSHTTPAnalyzer.UPSTREAM_KEY; @@ -71,7 +70,7 @@ public class MetaExchangeTCPAccessLogAnalyzer extends AbstractTCPAccessLogAnalyz final TCPAccessLogEntry entry, final Role role ) { - if (previousResult.hasResult()) { + if (previousResult.hasUpstreamMetrics() && previousResult.hasDownstreamMetrics()) { return previousResult; } if (!entry.hasCommonProperties()) { @@ -86,12 +85,14 @@ public class MetaExchangeTCPAccessLogAnalyzer extends AbstractTCPAccessLogAnalyz } final AccessLogCommon properties = entry.getCommonProperties(); final Map stateMap = properties.getFilterStateObjectsMap(); + final var newResult = previousResult.toBuilder(); + final var previousMetrics = previousResult.getMetrics(); if (stateMap.isEmpty()) { - return Result.builder().service(currSvc).build(); + return newResult.service(currSvc).build(); } - final TCPServiceMeshMetrics.Builder result = TCPServiceMeshMetrics.newBuilder(); - final AtomicBoolean downstreamExists = new AtomicBoolean(); + final var tcpMetrics = previousMetrics.getTcpMetricsBuilder(); + final var downstreamExists = new AtomicBoolean(); stateMap.forEach((key, value) -> { if (!key.equals(UPSTREAM_KEY) && !key.equals(DOWNSTREAM_KEY)) { return; @@ -106,19 +107,27 @@ public class MetaExchangeTCPAccessLogAnalyzer extends AbstractTCPAccessLogAnalyz final TCPServiceMeshMetric.Builder metrics; switch (key) { case UPSTREAM_KEY: + if (previousResult.hasUpstreamMetrics()) { + break; + } metrics = newAdapter(entry, currSvc, svc).adaptToUpstreamMetrics().setTlsMode(NON_TLS); if (log.isDebugEnabled()) { log.debug("Transformed a {} outbound mesh metrics {}", role, TextFormat.shortDebugString(metrics)); } - result.addMetrics(metrics); + tcpMetrics.addMetrics(metrics); + newResult.hasUpstreamMetrics(true); break; case DOWNSTREAM_KEY: + if (previousResult.hasDownstreamMetrics()) { + break; + } metrics = newAdapter(entry, svc, currSvc).adaptToDownstreamMetrics(); if (log.isDebugEnabled()) { log.debug("Transformed a {} inbound mesh metrics {}", role, TextFormat.shortDebugString(metrics)); } - result.addMetrics(metrics); + tcpMetrics.addMetrics(metrics); downstreamExists.set(true); + newResult.hasDownstreamMetrics(true); break; } }); @@ -127,9 +136,9 @@ public class MetaExchangeTCPAccessLogAnalyzer extends AbstractTCPAccessLogAnalyz if (log.isDebugEnabled()) { log.debug("Transformed a {} inbound mesh metric {}", role, TextFormat.shortDebugString(metric)); } - result.addMetrics(metric); + tcpMetrics.addMetrics(metric); } - return Result.builder().metrics(ServiceMeshMetrics.newBuilder().setTcpMetrics(result)).service(currSvc).build(); + return newResult.metrics(previousMetrics.setTcpMetrics(tcpMetrics)).service(currSvc).build(); } protected ServiceMetaInfo adaptToServiceMetaInfo(final Any value) throws Exception {