Parse TLS mode from ALS (#5454)

* Parse tls mode from als
Signed-off-by: NGao Hongtao <hanahmily@gmail.com>

* Polish code
Signed-off-by: NGao Hongtao <hanahmily@gmail.com>

proto @ 9933e2d1

-Subproject commit cdd58617e720949f51c0ddf5adf10b2b188e94fe
+Subproject commit 9933e2d17078c2bf07cd1c8d5ef36d52b5cbb917

docs/en/concepts-and-designs/scope-definitions.md

@@ -133,6 +133,7 @@ Calculate the metrics data from each request between one service and the other s
 | responseCode | Represent the response code of HTTP response, if this request is the HTTP call. | | int |
 | type | Represent the type of each request. Such as: Database, HTTP, RPC, gRPC. | | enum |
 | detectPoint | Represent where is the relation detected. Values: client, server, proxy. | yes | enum|
+| tlsMode | Represent TLS mode between source and destination services. For example `service_relation_mtls_cpm = from(ServiceRelation.*).filter(tlsMode == "mTLS").cpm()` || string|
 
 
 ### SCOPE `ServiceInstanceRelation`
@@ -154,6 +155,7 @@ Calculate the metrics data from each request between one service instance and th
 | responseCode | Represent the response code of HTTP response, if this request is the HTTP call. | | int |
 | type | Represent the type of each request. Such as: Database, HTTP, RPC, gRPC. | | enum |
 | detectPoint | Represent where is the relation detected. Values: client, server, proxy. | yes | enum|
+| tlsMode | Represent TLS mode between source and destination service instances. For example, `service_instance_relation_mtls_cpm = from(ServiceInstanceRelation.*).filter(tlsMode == "mTLS").cpm()` || string|
 
 ### SCOPE `EndpointRelation`
 

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/ServiceInstanceRelation.java

@@ -103,6 +103,9 @@ public class ServiceInstanceRelation extends Source {
     @Getter
     @Setter
     private DetectPoint detectPoint;
+    @Getter
+    @Setter
+    private String tlsMode;
 
     @Override
     public void prepare() {

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/source/ServiceRelation.java

@@ -93,6 +93,9 @@ public class ServiceRelation extends Source {
     @Getter
     @Setter
     private DetectPoint detectPoint;
+    @Getter
+    @Setter
+    private String tlsMode;
 
     @Override
     public void prepare() {

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

@@ -18,6 +18,7 @@
 
 package org.apache.skywalking.oap.server.receiver.envoy.als;
 
+import com.google.common.base.Strings;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import com.google.protobuf.Duration;
 import com.google.protobuf.Timestamp;
@@ -28,6 +29,7 @@ import io.envoyproxy.envoy.data.accesslog.v2.AccessLogCommon;
 import io.envoyproxy.envoy.data.accesslog.v2.HTTPAccessLogEntry;
 import io.envoyproxy.envoy.data.accesslog.v2.HTTPRequestProperties;
 import io.envoyproxy.envoy.data.accesslog.v2.HTTPResponseProperties;
+import io.envoyproxy.envoy.data.accesslog.v2.TLSProperties;
 import io.envoyproxy.envoy.service.accesslog.v2.StreamAccessLogsMessage;
 import io.kubernetes.client.openapi.ApiClient;
 import io.kubernetes.client.openapi.apis.CoreV1Api;
@@ -43,6 +45,7 @@ import java.util.Collections;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
@@ -68,6 +71,12 @@ public class K8sALSServiceMeshHTTPAnalysis implements ALSHTTPAnalysis {
 
     private static final String VALID_PHASE = "Running";
 
+    private static final String NON_TLS = "NONE";
+
+    private static final String M_TLS = "mTLS";
+
+    private static final String TLS = "TLS";
+
     @Getter(AccessLevel.PROTECTED)
     private final AtomicReference<Map<String, ServiceMetaInfo>> ipServiceMap = new AtomicReference<>();
 
@@ -211,6 +220,7 @@ public class K8sALSServiceMeshHTTPAnalysis implements ALSHTTPAnalysis {
                     downstreamLocalAddress.getSocketAddress()
                                           .getAddress(), downstreamLocalAddress.getSocketAddress()
                                                                                .getPortValue());
+                String tlsMode = parseTLS(properties.getTlsProperties());
                 if (cluster.startsWith("inbound|")) {
                     // Server side
                     if (downstreamService.equals(ServiceMetaInfo.UNKNOWN)) {
@@ -229,6 +239,7 @@ public class K8sALSServiceMeshHTTPAnalysis implements ALSHTTPAnalysis {
                                                                                 Math.toIntExact(responseCode))
                                                                             .setStatus(status)
                                                                             .setProtocol(protocol)
+                                                                            .setTlsMode(tlsMode)
                                                                             .setDetectPoint(DetectPoint.server);
 
                         LOGGER.debug("Transformed ingress->sidecar inbound mesh metric {}", metric);
@@ -252,6 +263,7 @@ public class K8sALSServiceMeshHTTPAnalysis implements ALSHTTPAnalysis {
                                                                                 Math.toIntExact(responseCode))
                                                                             .setStatus(status)
                                                                             .setProtocol(protocol)
+                                                                            .setTlsMode(tlsMode)
                                                                             .setDetectPoint(DetectPoint.server);
 
                         LOGGER.debug("Transformed sidecar->sidecar(server side) inbound mesh metric {}", metric);
@@ -281,6 +293,7 @@ public class K8sALSServiceMeshHTTPAnalysis implements ALSHTTPAnalysis {
                                                                         .setResponseCode(Math.toIntExact(responseCode))
                                                                         .setStatus(status)
                                                                         .setProtocol(protocol)
+                                                                        .setTlsMode(tlsMode)
                                                                         .setDetectPoint(DetectPoint.client);
 
                     LOGGER.debug("Transformed sidecar->sidecar(server side) inbound mesh metric {}", metric);
@@ -292,6 +305,21 @@ public class K8sALSServiceMeshHTTPAnalysis implements ALSHTTPAnalysis {
         return sources;
     }
 
+    private String parseTLS(TLSProperties properties) {
+        if (properties == null) {
+            return NON_TLS;
+        }
+        if (Strings.isNullOrEmpty(Optional.ofNullable(properties.getLocalCertificateProperties())
+            .orElse(TLSProperties.CertificateProperties.newBuilder().build()).getSubject())) {
+            return NON_TLS;
+        }
+        if (Strings.isNullOrEmpty(Optional.ofNullable(properties.getPeerCertificateProperties())
+            .orElse(TLSProperties.CertificateProperties.newBuilder().build()).getSubject())) {
+            return TLS;
+        }
+        return M_TLS;
+    }
+
     protected void analysisProxy(StreamAccessLogsMessage.Identifier identifier, HTTPAccessLogEntry entry) {
         AccessLogCommon properties = entry.getCommonProperties();
         if (properties != null) {
@@ -330,6 +358,7 @@ public class K8sALSServiceMeshHTTPAnalysis implements ALSHTTPAnalysis {
                     responseCode = response.getResponseCode().getValue();
                 }
                 boolean status = responseCode >= 200 && responseCode < 400;
+                String tlsMode = parseTLS(properties.getTlsProperties());
 
                 ServiceMeshMetric.Builder metric = ServiceMeshMetric.newBuilder()
                                                                     .setStartTime(startTime)
@@ -345,6 +374,7 @@ public class K8sALSServiceMeshHTTPAnalysis implements ALSHTTPAnalysis {
                                                                     .setResponseCode(Math.toIntExact(responseCode))
                                                                     .setStatus(status)
                                                                     .setProtocol(protocol)
+                                                                    .setTlsMode(tlsMode)
                                                                     .setDetectPoint(DetectPoint.server);
 
                 LOGGER.debug("Transformed ingress inbound mesh metric {}", metric);
@@ -376,6 +406,9 @@ public class K8sALSServiceMeshHTTPAnalysis implements ALSHTTPAnalysis {
                                                                                 Math.toIntExact(responseCode))
                                                                             .setStatus(status)
                                                                             .setProtocol(protocol)
+                                                                            // Can't parse it from tls properties, leave
+                                                                            // it to Server side.
+                                                                            .setTlsMode(NON_TLS)
                                                                             .setDetectPoint(DetectPoint.client);
 
                 LOGGER.debug("Transformed ingress outbound mesh metric {}", outboundMetric);

oap-server/server-receiver-plugin/skywalking-mesh-receiver-plugin/src/main/java/org/apache/skywalking/aop/server/receiver/mesh/TelemetryDataDispatcher.java

@@ -186,6 +186,7 @@ public class TelemetryDataDispatcher {
         serviceRelation.setResponseCode(metrics.getResponseCode());
         serviceRelation.setDetectPoint(detectPointMapping(metrics.getDetectPoint()));
         serviceRelation.setComponentId(protocol2Component(metrics.getProtocol()));
+        serviceRelation.setTlsMode(metrics.getTlsMode());
 
         SOURCE_RECEIVER.receive(serviceRelation);
     }
@@ -221,6 +222,7 @@ public class TelemetryDataDispatcher {
         serviceRelation.setResponseCode(metrics.getResponseCode());
         serviceRelation.setDetectPoint(detectPointMapping(metrics.getDetectPoint()));
         serviceRelation.setComponentId(protocol2Component(metrics.getProtocol()));
+        serviceRelation.setTlsMode(metrics.getTlsMode());
 
         SOURCE_RECEIVER.receive(serviceRelation);
     }