Fix NPE and open some methods for extensibility (#5832)

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

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<String/* ip */, ServiceMetaInfo> ipServiceMetaInfoMap;
+public class K8SServiceRegistry {
+    protected final Map<String/* ip */, ServiceMetaInfo> ipServiceMetaInfoMap;
 
-    final Map<String/* namespace:serviceName */, V1Service> idServiceMap;
+    protected final Map<String/* namespace:serviceName */, V1Service> idServiceMap;
 
-    final Map<String/* ip */, V1Pod> ipPodMap;
+    protected final Map<String/* ip */, V1Pod> ipPodMap;
 
-    final Map<String/* ip */, String/* namespace:serviceName */> ipServiceMap;
+    protected final Map<String/* ip */, String/* namespace:serviceName */> 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<ServiceMetaInfo.KeyValue> transformLabelsToTags(final Map<String, String> labels) {
+    protected List<ServiceMetaInfo.KeyValue> transformLabelsToTags(final Map<String, String> 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();
     }
 }

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

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<String> 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<String, Object> context) throws Exception {
+    public String format(final Map<String, Object> context) throws Exception {
         final Object[] values = new Object[properties.size()];
 
         for (int i = 0; i < properties.size(); i++) {

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<String, Invokable<ServiceMetaInfo, ?>> fieldSetterMapping;
+    private Map<String, Method> 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<? extends ServiceMetaInfo> 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<? extends ServiceMetaInfo> serviceInfoClass) throws ModuleStartException {
         if (initialized) {
             return;
         }
+
         final Yaml yaml = new Yaml();
         final Map<String, String> config = (Map<String, String>) yaml.load(inputStream);
 
@@ -92,11 +96,9 @@ enum FieldsHelper {
             );
 
             try {
-                final Method setterMethod = ServiceMetaInfo.class.getMethod("set" + StringUtils.capitalize(serviceMetaInfoFieldName), String.class);
-                final Invokable<ServiceMetaInfo, ?> setter = new TypeToken<ServiceMetaInfo>() {
-                }.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<String, ServiceNameFormat> entry : fieldNameMapping.entrySet()) {
             final ServiceNameFormat serviceNameFormat = entry.getValue();
@@ -120,7 +123,7 @@ enum FieldsHelper {
                 final List<String> 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();
             }

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());
+    }
+
 }

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