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

Check results in ALS as per downstream/upstream instead of per log (#11265)

This commit is contained in:
kezhenxu94 2023-08-27 12:20:35 +08:00 committed by GitHub
parent ecf62c5dbc
commit 16f940b5ce
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
6 changed files with 148 additions and 93 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

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

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

@ -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<E> {
/**
* 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.
*
* <p>
* 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<E> {
}
@Data
@Builder
static class Result {
@Builder(toBuilder = true)
class Result {
/**
* The service representing the Envoy node.
*/
@ -78,10 +80,16 @@ public interface AccessLogAnalyzer<E> {
/**
* 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;
}
}

82
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

@ -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();
}
/**

36
oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/mx/MetaExchangeALSHTTPAnalyzer.java
View File

@ -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<String, Any> 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 {

65
oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/tcp/k8s/K8sALSServiceMeshTCPAnalysis.java
View File

@ -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();
}
/**

35
oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/tcp/mx/MetaExchangeTCPAccessLogAnalyzer.java
View File

@ -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<String, Any> 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 {