Added Ztunnel component in the Topology (#13251)

Author: mrproliu

docs/en/changes/changes.md

@@ -17,6 +17,8 @@
 * Increase the idle check interval of the message queue to 200ms to reduce CPU usage under low load conditions.
 * Limit max attempts of DNS resolution of Istio ServiceEntry to 3, and do not wait for first resolution result in case the DNS is not resolvable at all.
 * Support analysis waypoint metrics in Envoy ALS receiver.
+* Add Ztunnel component in the topology.
+* [Break Change] Change `compomentId` to `componentIds` in the K8SServiceRelation Scope. 
 
 #### UI
 

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

@@ -493,15 +493,15 @@ following **package**/**protocol** level metric contents.
 
 ##### SCOPE `K8SServiceRelation`
 
-| Name              | Remarks                                                            | Group Key | Type   |
-|-------------------|--------------------------------------------------------------------|-----------|--------|
-| sourceServiceName | The source service name in kubernetes.                             |           | string |
-| sourceLayer       | The source layer service in kubernetes.                            |           | string |
-| detectPoint       | Where the relation is detected. The value may be client or server. |           | enum   |
-| componentId       | The ID of component used in this call.                             |           | string |
-| tlsMode           | The TLS mode of relation. The value may be Plain or TLS.           |           | enum   |
-| destServiceName   | The dest service name in kubernetes.                               |           | string |
-| destLayer         | The dest layer service in kubernetes.                              |           | string |
+| Name              | Remarks                                                            | Group Key | Type         |
+|-------------------|--------------------------------------------------------------------|-----------|--------------|
+| sourceServiceName | The source service name in kubernetes.                             |           | string       |
+| sourceLayer       | The source layer service in kubernetes.                            |           | string       |
+| detectPoint       | Where the relation is detected. The value may be client or server. |           | enum         |
+| componentIds      | The ID of component used in this call.                             |           | List<string> |
+| tlsMode           | The TLS mode of relation. The value may be Plain or TLS.           |           | enum         |
+| destServiceName   | The dest service name in kubernetes.                               |           | string       |
+| destLayer         | The dest layer service in kubernetes.                              |           | string       |
 
 ##### SCOPE `K8SServiceRelation`
 

oap-server/oal-rt/src/test/java/org/apache/skywalking/oal/rt/parser/ScriptParserTest.java

@@ -326,6 +326,25 @@ public void testParse12() throws IOException {
         Assertions.assertEquals("(str->long)", castExp.getAggregationFuncStmt().getFuncArgs().get(0).getCastType());
     }
 
+    @Test
+    public void testParse13() throws IOException {
+        ScriptParser parser = ScriptParser.createFromScriptText(
+            "ClientCpm = from(K8SServiceRelation.*).filter(componentIds contain 7).cpm();", TEST_SOURCE_PACKAGE);
+        List<AnalysisResult> results = parser.parse().getMetricsStmts();
+        AnalysisResult clientCpm = results.get(0);
+        Assertions.assertEquals("ClientCpm", clientCpm.getMetricsName());
+        Assertions.assertEquals("K8SServiceRelation", clientCpm.getFrom().getSourceName());
+        Assertions.assertEquals("[*]", clientCpm.getFrom().getSourceAttribute().toString());
+        final List<Expression> filterExpressions = clientCpm.getFilters().getFilterExpressions();
+        Assertions.assertEquals(1, filterExpressions.size());
+        Assertions.assertEquals("source.getComponentIds()", filterExpressions.get(0).getLeft());
+        Assertions.assertEquals("7", filterExpressions.get(0).getRight());
+        Assertions.assertEquals("cpm", clientCpm.getAggregationFuncStmt().getAggregationFunctionName());
+        EntryMethod entryMethod = clientCpm.getEntryMethod();
+        List<Object> methodArgsExpressions = entryMethod.getArgsExpressions();
+        Assertions.assertEquals(1, methodArgsExpressions.size());
+    }
+
     @Test
     public void testParseDecorator() throws IOException {
         SourceDecoratorManager.DECORATOR_MAP.put("ServiceDecorator", new ISourceDecorator<ISource>() {

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/manual/relation/service/K8SServiceCallRelationDispatcher.java

@@ -42,7 +42,7 @@ private void clientSide(K8SServiceRelation source) {
         metrics.setSourceServiceId(source.getSourceServiceId());
         metrics.setDestServiceId(source.getDestServiceId());
         final IntList componentIds = metrics.getComponentIds();
-        componentIds.add(source.getComponentId());
+        source.getComponentIds().forEach(componentIds::add);
         metrics.setEntityId(source.getEntityId());
         MetricsStreamProcessor.getInstance().in(metrics);
     }
@@ -53,7 +53,7 @@ private void serverSide(K8SServiceRelation source) {
         metrics.setSourceServiceId(source.getSourceServiceId());
         metrics.setDestServiceId(source.getDestServiceId());
         final IntList componentIds = metrics.getComponentIds();
-        componentIds.add(source.getComponentId());
+        source.getComponentIds().forEach(componentIds::add);
         metrics.setEntityId(source.getEntityId());
         MetricsStreamProcessor.getInstance().in(metrics);
     }

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/metrics/expression/ContainMatch.java

@@ -33,4 +33,16 @@ public boolean match(List<String> left, String right) {
         }
         return left.contains(right);
     }
+
+    public boolean match(List<Integer> left, Integer right) {
+        if (Objects.isNull(left)) {
+            return false;
+        }
+        return left.contains(right);
+    }
+
+    public boolean match(List<Integer> left, int right) {
+        return match(left, Integer.valueOf(right));
+    }
+
 }

oap-server/server-core/src/main/java/org/apache/skywalking/oap/server/core/analysis/metrics/expression/NotContainMatch.java

@@ -33,4 +33,15 @@ public boolean match(List<String> left, String right) {
         }
         return !left.contains(right);
     }
+
+    public boolean match(List<Integer> left, Integer right) {
+        if (Objects.isNull(left)) {
+            return false;
+        }
+        return !left.contains(right);
+    }
+
+    public boolean match(List<Integer> left, int right) {
+        return match(left, Integer.valueOf(right));
+    }
 }

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

@@ -23,6 +23,9 @@
 import org.apache.skywalking.oap.server.core.analysis.IDManager;
 import org.apache.skywalking.oap.server.core.analysis.Layer;
 
+import java.util.ArrayList;
+import java.util.List;
+
 import static org.apache.skywalking.oap.server.core.source.DefaultScopeDefine.K8S_SERVICE_RELATION;
 import static org.apache.skywalking.oap.server.core.source.DefaultScopeDefine.SERVICE_RELATION_CATALOG_NAME;
 
@@ -39,7 +42,7 @@ public class K8SServiceRelation extends K8SMetrics {
     private Layer sourceLayer;
 
     private DetectPoint detectPoint;
-    private int componentId;
+    private final List<Integer> componentIds = new ArrayList<>(3);
     private AccessLogConnectionTLSMode tlsMode;
 
     private String destServiceId;

oap-server/server-core/src/test/java/org/apache/skywalking/oap/server/core/analysis/metrics/expression/ContainMatchTest.java

@@ -33,5 +33,7 @@ public void match() {
         assertTrue(containMatch.match(Arrays.asList("http.method:GET", "http.method:POST"), "http.method:GET"));
         assertFalse(
             containMatch.match(Arrays.asList("http.method:GET", "http.method:POST"), "http.method:PUT"));
+        assertTrue(containMatch.match(Arrays.asList(1, 2, 3), 2));
+        assertFalse(containMatch.match(Arrays.asList(1, 2, 3), 4));
     }
 }

oap-server/server-receiver-plugin/skywalking-ebpf-receiver-plugin/src/main/java/org/apache/skywalking/oap/server/receiver/ebpf/provider/handler/AccessLogServiceHandler.java

@@ -533,13 +533,13 @@ protected KubernetesProcessAddress buildRemoteAddress(NodeInfo nodeInfo, ObjectI
             .build();
     }
 
-    protected int buildConnectionComponentId(ConnectionInfo connectionInfo) {
+    protected List<Integer> buildConnectionComponentId(ConnectionInfo connectionInfo) {
         final AccessLogConnection originalConnection = connectionInfo.getOriginalConnection();
         if (originalConnection.hasAttachment() && originalConnection.getAttachment().hasZTunnel() &&
             ZTunnelAttachmentSecurityPolicy.MTLS.equals(originalConnection.getAttachment().getZTunnel().getSecurityPolicy())) {
-            return 142; // mTLS
+            return Arrays.asList(142, 162); // mTLS, ztunnel
         }
-        return buildProtocolComponentID(connectionInfo);
+        return Arrays.asList(buildProtocolComponentID(connectionInfo));
     }
 
     protected int buildProtocolComponentID(ConnectionInfo connectionInfo) {
@@ -653,7 +653,7 @@ public K8SServiceRelation toServiceRelation() {
             serviceRelation.setSourceLayer(Layer.K8S_SERVICE);
 
             serviceRelation.setDetectPoint(parseToSourceRole());
-            serviceRelation.setComponentId(buildConnectionComponentId(this));
+            serviceRelation.getComponentIds().addAll(buildConnectionComponentId(this));
             serviceRelation.setTlsMode(tlsMode);
 
             serviceRelation.setDestServiceName(destServiceName);

oap-server/server-starter/src/main/resources/component-libraries.yml

@@ -535,6 +535,9 @@ Caffeine:
 ThreadPerTask-executor:
   id: 161
   languages: Java
+ztunnel:
+  id: 162
+  languages: ebpf, mesh
 
 # .NET/.NET Core components
 # [3000, 4000) for C#/.NET only