diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/LogEntry2MetricsAdapter.java b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/LogEntry2MetricsAdapter.java index de05c7a26..b3dee9ff3 100644 --- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/LogEntry2MetricsAdapter.java +++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/LogEntry2MetricsAdapter.java @@ -51,11 +51,11 @@ public class LogEntry2MetricsAdapter { /** * The access log entry that is to be adapted into metrics builders. */ - private final HTTPAccessLogEntry entry; + protected final HTTPAccessLogEntry entry; - private final ServiceMetaInfo sourceService; + protected final ServiceMetaInfo sourceService; - private final ServiceMetaInfo targetService; + protected final ServiceMetaInfo targetService; /** * Adapt the {@code entry} into a downstream metrics {@link ServiceMeshMetric.Builder}. @@ -94,7 +94,7 @@ public class LogEntry2MetricsAdapter { protected ServiceMeshMetric.Builder adaptCommonPart() { final AccessLogCommon properties = entry.getCommonProperties(); - final String endpoint = ofNullable(entry.getRequest()).map(HTTPRequestProperties::getPath).orElse("/"); + final String endpoint = endpoint(); final int responseCode = ofNullable(entry.getResponse()).map(HTTPResponseProperties::getResponseCode).map(UInt32Value::getValue).orElse(200); final boolean status = responseCode >= 200 && responseCode < 400; final Protocol protocol = requestProtocol(entry.getRequest()); @@ -124,6 +124,10 @@ public class LogEntry2MetricsAdapter { return builder; } + protected String endpoint() { + return ofNullable(entry.getRequest()).map(HTTPRequestProperties::getPath).orElse("/"); + } + protected static long formatAsLong(final Timestamp timestamp) { return Instant.ofEpochSecond(timestamp.getSeconds(), timestamp.getNanos()).toEpochMilli(); } diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/k8s/K8SServiceRegistry.java b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/k8s/K8SServiceRegistry.java index 1f116e982..5ecb37947 100644 --- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/k8s/K8SServiceRegistry.java +++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/k8s/K8SServiceRegistry.java @@ -52,20 +52,20 @@ import static java.util.Objects.isNull; import static java.util.Objects.requireNonNull; @Slf4j -class K8SServiceRegistry { - final Map ipServiceMetaInfoMap; +public class K8SServiceRegistry { + protected final Map ipServiceMetaInfoMap; - final Map idServiceMap; + protected final Map idServiceMap; - final Map ipPodMap; + protected final Map ipPodMap; - final Map ipServiceMap; + protected final Map ipServiceMap; - final ExecutorService executor; + protected final ExecutorService executor; - final ServiceNameFormatter serviceNameFormatter; + protected final ServiceNameFormatter serviceNameFormatter; - K8SServiceRegistry(final EnvoyMetricReceiverConfig config) { + public K8SServiceRegistry(final EnvoyMetricReceiverConfig config) { serviceNameFormatter = new ServiceNameFormatter(config.getK8sServiceNameRule()); ipServiceMetaInfoMap = new ConcurrentHashMap<>(); idServiceMap = new ConcurrentHashMap<>(); @@ -79,7 +79,7 @@ class K8SServiceRegistry { ); } - void start() throws IOException { + public void start() throws IOException { final ApiClient apiClient = Config.defaultClient(); apiClient.setHttpClient(apiClient.getHttpClient() .newBuilder() @@ -200,7 +200,7 @@ class K8SServiceRegistry { }); } - private void addService(final V1Service service) { + protected void addService(final V1Service service) { Optional.ofNullable(service.getMetadata()).ifPresent( metadata -> idServiceMap.put(metadata.getNamespace() + ":" + metadata.getName(), service) ); @@ -208,13 +208,13 @@ class K8SServiceRegistry { recompose(); } - private void removeService(final V1Service service) { + protected void removeService(final V1Service service) { Optional.ofNullable(service.getMetadata()).ifPresent( metadata -> idServiceMap.remove(metadata.getUid()) ); } - private void addPod(final V1Pod pod) { + protected void addPod(final V1Pod pod) { Optional.ofNullable(pod.getStatus()).ifPresent( status -> ipPodMap.put(status.getPodIP(), pod) ); @@ -222,13 +222,13 @@ class K8SServiceRegistry { recompose(); } - private void removePod(final V1Pod pod) { + protected void removePod(final V1Pod pod) { Optional.ofNullable(pod.getStatus()).ifPresent( status -> ipPodMap.remove(status.getPodIP()) ); } - private void addEndpoints(final V1Endpoints endpoints) { + protected void addEndpoints(final V1Endpoints endpoints) { final String namespace = requireNonNull(endpoints.getMetadata()).getNamespace(); final String name = requireNonNull(endpoints.getMetadata()).getName(); @@ -241,7 +241,7 @@ class K8SServiceRegistry { recompose(); } - private void removeEndpoints(final V1Endpoints endpoints) { + protected void removeEndpoints(final V1Endpoints endpoints) { requireNonNull(endpoints.getSubsets()).forEach( subset -> requireNonNull(subset.getAddresses()).forEach( address -> ipServiceMap.remove(address.getIp()) @@ -249,7 +249,7 @@ class K8SServiceRegistry { ); } - private List transformLabelsToTags(final Map labels) { + protected List transformLabelsToTags(final Map labels) { if (isNull(labels)) { return Collections.emptyList(); } @@ -259,7 +259,7 @@ class K8SServiceRegistry { .collect(Collectors.toList()); } - ServiceMetaInfo findService(final String ip) { + protected ServiceMetaInfo findService(final String ip) { final ServiceMetaInfo service = ipServiceMetaInfoMap.get(ip); if (isNull(service)) { log.debug("Unknown ip {}, ip -> service is null", ip); @@ -268,7 +268,7 @@ class K8SServiceRegistry { return service; } - private void recompose() { + protected void recompose() { ipPodMap.forEach((ip, pod) -> { final String namespaceService = ipServiceMap.get(ip); final V1Service service; @@ -297,7 +297,7 @@ class K8SServiceRegistry { }); } - boolean isEmpty() { + protected boolean isEmpty() { return ipServiceMetaInfoMap.isEmpty(); } } 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 0e2e0682f..357f6c707 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 @@ -127,7 +127,8 @@ public class K8sALSServiceMeshHTTPAnalysis extends AbstractALSAnalyzer { return Collections.emptyList(); } final Address downstreamLocalAddress = properties.getDownstreamLocalAddress(); - final Address downstreamRemoteAddress = properties.getDownstreamRemoteAddress(); + final Address downstreamRemoteAddress = properties.hasDownstreamDirectRemoteAddress() ? + properties.getDownstreamDirectRemoteAddress() : properties.getDownstreamRemoteAddress(); final Address upstreamRemoteAddress = properties.getUpstreamRemoteAddress(); if (downstreamLocalAddress == null || downstreamRemoteAddress == null || upstreamRemoteAddress == null) { return Collections.emptyList(); diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/k8s/ServiceNameFormatter.java b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/k8s/ServiceNameFormatter.java index 5e2e37508..12534de0e 100644 --- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/k8s/ServiceNameFormatter.java +++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/k8s/ServiceNameFormatter.java @@ -27,13 +27,13 @@ import java.util.regex.Pattern; import org.apache.commons.beanutils.PropertyUtils; import org.apache.commons.lang3.StringUtils; -class ServiceNameFormatter { +public class ServiceNameFormatter { private final List properties; private final StringBuffer serviceNamePattern; - ServiceNameFormatter(String rule) { + public ServiceNameFormatter(String rule) { rule = StringUtils.defaultIfBlank(rule, "${service.metadata.name}"); this.properties = new ArrayList<>(); @@ -48,7 +48,7 @@ class ServiceNameFormatter { } } - String format(final Map context) throws Exception { + public String format(final Map context) throws Exception { final Object[] values = new Object[properties.size()]; for (int i = 0; i < properties.size(); i++) { diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/mx/FieldsHelper.java b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/mx/FieldsHelper.java index 1ef5c4551..f3d27240b 100644 --- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/mx/FieldsHelper.java +++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/mx/FieldsHelper.java @@ -20,8 +20,6 @@ package org.apache.skywalking.oap.server.receiver.envoy.als.mx; import com.google.common.base.Splitter; import com.google.common.base.Strings; -import com.google.common.reflect.Invokable; -import com.google.common.reflect.TypeToken; import com.google.protobuf.Struct; import com.google.protobuf.Value; import java.io.InputStream; @@ -41,8 +39,7 @@ import org.apache.skywalking.oap.server.receiver.envoy.als.ServiceMetaInfo; import org.yaml.snakeyaml.Yaml; @Slf4j -@SuppressWarnings("UnstableApiUsage") -enum FieldsHelper { +public enum FieldsHelper { SINGLETON; private boolean initialized = false; @@ -55,17 +52,24 @@ enum FieldsHelper { /** * The mappings from the field name of {@link ServiceMetaInfo} to its {@code setter}. */ - private Map> fieldSetterMapping; + private Map fieldSetterMapping; public void init(final String file) throws Exception { - init(ResourceUtils.readToStream(file)); + init(ResourceUtils.readToStream(file), ServiceMetaInfo.class); + } + + public void init(final String file, + final Class serviceInfoClass) throws Exception { + init(ResourceUtils.readToStream(file), serviceInfoClass); } @SuppressWarnings("unchecked") - void init(final InputStream inputStream) throws ModuleStartException { + public void init(final InputStream inputStream, + final Class serviceInfoClass) throws ModuleStartException { if (initialized) { return; } + final Yaml yaml = new Yaml(); final Map config = (Map) yaml.load(inputStream); @@ -92,11 +96,9 @@ enum FieldsHelper { ); try { - final Method setterMethod = ServiceMetaInfo.class.getMethod("set" + StringUtils.capitalize(serviceMetaInfoFieldName), String.class); - final Invokable setter = new TypeToken() { - }.method(setterMethod); - setter.setAccessible(true); - fieldSetterMapping.put(serviceMetaInfoFieldName, setter); + final Method setterMethod = serviceInfoClass.getMethod("set" + StringUtils.capitalize(serviceMetaInfoFieldName), String.class); + setterMethod.setAccessible(true); + fieldSetterMapping.put(serviceMetaInfoFieldName, setterMethod); } catch (final NoSuchMethodException e) { throw new ModuleStartException("Initialize method error", e); } @@ -112,6 +114,7 @@ enum FieldsHelper { * @throws Exception if failed to inflate the {@code serviceMetaInfo} */ public void inflate(final Struct metadata, final ServiceMetaInfo serviceMetaInfo) throws Exception { + final Value empty = Value.newBuilder().setStringValue("-").build(); final Value root = Value.newBuilder().setStructValue(metadata).build(); for (final Map.Entry entry : fieldNameMapping.entrySet()) { final ServiceNameFormat serviceNameFormat = entry.getValue(); @@ -120,7 +123,7 @@ enum FieldsHelper { final List properties = serviceNameFormat.properties.get(i); Value value = root; for (final String property : properties) { - value = value.getStructValue().getFieldsOrThrow(property); + value = value.getStructValue().getFieldsOrDefault(property, empty); } values[i] = value.getStringValue(); } 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 796c23e74..741a203e1 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 @@ -44,9 +44,11 @@ import static org.apache.skywalking.oap.server.receiver.envoy.als.ServiceMetaInf @Slf4j public class MetaExchangeALSHTTPAnalyzer extends AbstractALSAnalyzer { - private static final String UPSTREAM_KEY = "wasm.upstream_peer"; + public static final String UPSTREAM_KEY = "wasm.upstream_peer"; - private static final String DOWNSTREAM_KEY = "wasm.downstream_peer"; + public static final String DOWNSTREAM_KEY = "wasm.downstream_peer"; + + protected String fieldMappingFile = "metadata-service-mapping.yaml"; @Override public String name() { @@ -56,7 +58,7 @@ public class MetaExchangeALSHTTPAnalyzer extends AbstractALSAnalyzer { @Override public void init(ModuleManager manager, EnvoyMetricReceiverConfig config) throws ModuleStartException { try { - FieldsHelper.SINGLETON.init("metadata-service-mapping.yaml"); + FieldsHelper.SINGLETON.init(fieldMappingFile); } catch (final Exception e) { throw new ModuleStartException("Failed to load metadata-service-mapping.yaml", e); } @@ -74,7 +76,7 @@ public class MetaExchangeALSHTTPAnalyzer extends AbstractALSAnalyzer { } final ServiceMetaInfo currSvc; try { - currSvc = new ServiceMetaInfoAdapter(identifier.getNode().getMetadata()); + currSvc = adaptToServiceMetaInfo(identifier); } catch (Exception e) { log.error("Failed to inflate the ServiceMetaInfo from identifier.node.metadata. ", e); return Collections.emptyList(); @@ -88,7 +90,7 @@ public class MetaExchangeALSHTTPAnalyzer extends AbstractALSAnalyzer { } final ServiceMetaInfo svc; try { - svc = new ServiceMetaInfoAdapter(value); + svc = adaptToServiceMetaInfo(value); } catch (Exception e) { log.error("Fail to parse metadata {} to FlatNode", Base64.getEncoder().encode(value.toByteArray())); return; @@ -122,4 +124,12 @@ public class MetaExchangeALSHTTPAnalyzer extends AbstractALSAnalyzer { return result; } + protected ServiceMetaInfo adaptToServiceMetaInfo(final Any value) throws Exception { + return new ServiceMetaInfoAdapter(value); + } + + protected ServiceMetaInfo adaptToServiceMetaInfo(final StreamAccessLogsMessage.Identifier identifier) throws Exception { + return new ServiceMetaInfoAdapter(identifier.getNode().getMetadata()); + } + } diff --git a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/envoy/als/mx/FieldsHelperTest.java b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/envoy/als/mx/FieldsHelperTest.java index 35e91f858..40f327bc9 100644 --- a/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/envoy/als/mx/FieldsHelperTest.java +++ b/oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/test/java/org/apache/skywalking/oap/server/receiver/envoy/als/mx/FieldsHelperTest.java @@ -84,7 +84,7 @@ public class FieldsHelperTest { final StreamAccessLogsMessage.Builder requestBuilder = StreamAccessLogsMessage.newBuilder(); JsonFormat.parser().merge(isr, requestBuilder); final ServiceMetaInfo info = new ServiceMetaInfo(); - FieldsHelper.SINGLETON.init(new ByteArrayInputStream(mapping.getBytes())); + FieldsHelper.SINGLETON.init(new ByteArrayInputStream(mapping.getBytes()), ServiceMetaInfo.class); FieldsHelper.SINGLETON.inflate( requestBuilder.getIdentifier().getNode().getMetadata(), info