Add some defensive codes of new protobuffers' semantics (#6737)

CHANGES.md

@@ -11,6 +11,7 @@ Release Notes.
 
 
 #### OAP-Backend
+* BugFix: filter invalid Envoy access logs whose socket address is empty.
 * Fix K8s monitoring the incorrect metrics calculate. 
 
 #### UI

oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/LogEntry2MetricsAdapter.java

@@ -20,11 +20,9 @@ package org.apache.skywalking.oap.server.receiver.envoy.als;
 
 import com.google.protobuf.Duration;
 import com.google.protobuf.Timestamp;
-import com.google.protobuf.UInt32Value;
 import io.envoyproxy.envoy.data.accesslog.v3.AccessLogCommon;
 import io.envoyproxy.envoy.data.accesslog.v3.HTTPAccessLogEntry;
 import io.envoyproxy.envoy.data.accesslog.v3.HTTPRequestProperties;
-import io.envoyproxy.envoy.data.accesslog.v3.HTTPResponseProperties;
 import io.envoyproxy.envoy.data.accesslog.v3.ResponseFlags;
 import io.envoyproxy.envoy.data.accesslog.v3.TLSProperties;
 import java.time.Instant;
@@ -36,7 +34,6 @@ import org.apache.skywalking.apm.network.servicemesh.v3.Protocol;
 import org.apache.skywalking.apm.network.servicemesh.v3.ServiceMeshMetric;
 
 import static com.google.common.base.Strings.isNullOrEmpty;
-import static java.util.Optional.ofNullable;
 
 /**
  * Adapt {@link HTTPAccessLogEntry} objects to {@link ServiceMeshMetric} builders.
@@ -97,9 +94,8 @@ public class LogEntry2MetricsAdapter {
     protected ServiceMeshMetric.Builder adaptCommonPart() {
         final AccessLogCommon properties = entry.getCommonProperties();
         final String endpoint = endpoint();
-        final int responseCode = ofNullable(entry.getResponse()).map(HTTPResponseProperties::getResponseCode)
-                                                                .map(UInt32Value::getValue)
-                                                                .orElse(200);
+        int responseCode = entry.getResponse().getResponseCode().getValue();
+        responseCode = responseCode > 0 ? responseCode : 200;
         final boolean status = responseCode >= 200 && responseCode < 400;
         final Protocol protocol = requestProtocol(entry.getRequest());
         final String tlsMode = parseTLS(properties.getTlsProperties());
@@ -162,15 +158,11 @@ public class LogEntry2MetricsAdapter {
         if (properties == null) {
             return NON_TLS;
         }
-        TLSProperties.CertificateProperties lp = Optional
-            .ofNullable(properties.getLocalCertificateProperties())
-            .orElse(TLSProperties.CertificateProperties.newBuilder().build());
+        TLSProperties.CertificateProperties lp = properties.getLocalCertificateProperties();
         if (isNullOrEmpty(lp.getSubject()) && !hasSAN(lp.getSubjectAltNameList())) {
             return NON_TLS;
         }
-        TLSProperties.CertificateProperties pp = Optional
-            .ofNullable(properties.getPeerCertificateProperties())
-            .orElse(TLSProperties.CertificateProperties.newBuilder().build());
+        TLSProperties.CertificateProperties pp = properties.getPeerCertificateProperties();
         if (isNullOrEmpty(pp.getSubject()) && !hasSAN(pp.getSubjectAltNameList())) {
             return TLS;
         }

oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/k8s/Addresses.java

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.apache.skywalking.oap.server.receiver.envoy.als.k8s;
+
+import io.envoyproxy.envoy.config.core.v3.Address;
+
+import static java.util.Objects.nonNull;
+import static org.apache.skywalking.apm.util.StringUtil.isNotBlank;
+
+public class Addresses {
+    public static boolean isValid(final Address address) {
+        return nonNull(address)
+            && address.hasSocketAddress()
+            && isNotBlank(address.getSocketAddress().getAddress());
+    }
+}

oap-server/server-receiver-plugin/envoy-metrics-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/envoy/als/k8s/K8sALSServiceMeshHTTPAnalysis.java

@@ -35,8 +35,10 @@ 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 static org.apache.skywalking.apm.util.StringUtil.isBlank;
 import static org.apache.skywalking.oap.server.library.util.CollectionUtils.isNotEmpty;
 import static org.apache.skywalking.oap.server.receiver.envoy.als.LogEntry2MetricsAdapter.NON_TLS;
+import static org.apache.skywalking.oap.server.receiver.envoy.als.k8s.Addresses.isValid;
 
 /**
  * Analysis log based on ingress and mesh scenarios.
@@ -81,12 +83,12 @@ public class K8sALSServiceMeshHTTPAnalysis extends AbstractALSAnalyzer {
     }
 
     protected List<ServiceMeshMetric.Builder> analyzeSideCar(final HTTPAccessLogEntry entry) {
-        final AccessLogCommon properties = entry.getCommonProperties();
-        if (properties == null) {
+        if (!entry.hasCommonProperties()) {
             return Collections.emptyList();
         }
+        final AccessLogCommon properties = entry.getCommonProperties();
         final String cluster = properties.getUpstreamCluster();
-        if (cluster == null) {
+        if (isBlank(cluster)) {
             return Collections.emptyList();
         }
 
@@ -98,6 +100,9 @@ public class K8sALSServiceMeshHTTPAnalysis extends AbstractALSAnalyzer {
                 : properties.getDownstreamRemoteAddress();
         final ServiceMetaInfo downstreamService = find(downstreamRemoteAddress.getSocketAddress().getAddress());
         final Address downstreamLocalAddress = properties.getDownstreamLocalAddress();
+        if (!isValid(downstreamRemoteAddress) || !isValid(downstreamLocalAddress)) {
+            return Collections.emptyList();
+        }
         final ServiceMetaInfo localService = find(downstreamLocalAddress.getSocketAddress().getAddress());
 
         if (cluster.startsWith("inbound|")) {
@@ -119,6 +124,9 @@ public class K8sALSServiceMeshHTTPAnalysis extends AbstractALSAnalyzer {
         } else if (cluster.startsWith("outbound|")) {
             // sidecar(client side) -> sidecar
             final Address upstreamRemoteAddress = properties.getUpstreamRemoteAddress();
+            if (!isValid(upstreamRemoteAddress)) {
+                return sources;
+            }
             final ServiceMetaInfo destService = find(upstreamRemoteAddress.getSocketAddress().getAddress());
 
             final ServiceMeshMetric.Builder metric = newAdapter(entry, downstreamService, destService).adaptToUpstreamMetrics();
@@ -131,15 +139,15 @@ public class K8sALSServiceMeshHTTPAnalysis extends AbstractALSAnalyzer {
     }
 
     protected List<ServiceMeshMetric.Builder> analyzeProxy(final HTTPAccessLogEntry entry) {
-        final AccessLogCommon properties = entry.getCommonProperties();
-        if (properties == null) {
+        if (!entry.hasCommonProperties()) {
             return Collections.emptyList();
         }
+        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 (downstreamLocalAddress == null || downstreamRemoteAddress == null || upstreamRemoteAddress == null) {
+        if (!isValid(downstreamLocalAddress) || !isValid(downstreamRemoteAddress) || !isValid(upstreamRemoteAddress)) {
             return Collections.emptyList();
         }