Adding eBPF Access Log Feature E2E and support exclude specific names…

…paces traffic (#12304)
apache · Jun 6, 2024 · 0b175ff · 0b175ff
1 parent 341bdcf
commit 0b175ff
Show file tree

Hide file tree

Showing 18 changed files with 886 additions and 4 deletions.
diff --git a/.github/workflows/skywalking.yaml b/.github/workflows/skywalking.yaml
@@ -531,6 +531,14 @@ jobs:
               file: Dockerfile.sqrt
               name: test/continuous:test
 
+          # eBPF Access Log
+          - name: eBPF Access Log BanyanDB
+            config: test/e2e-v2/cases/profiling/ebpf/access_log/banyandb/e2e.yaml
+          - name: eBPF Access Log ES
+            config: test/e2e-v2/cases/profiling/ebpf/access_log/es/e2e.yaml
+          - name: eBPF Access Log ES Sharding
+            config: test/e2e-v2/cases/profiling/ebpf/access_log/es/es-sharding/e2e.yaml
+
           - name: Kafka Basic
             config: test/e2e-v2/cases/kafka/simple-so11y/e2e.yaml
           - name: Kafka Profiling

diff --git a/apm-protocol/apm-network/src/main/proto b/apm-protocol/apm-network/src/main/proto
diff --git a/docs/en/changes/changes.md b/docs/en/changes/changes.md
@@ -5,6 +5,7 @@
 
 #### OAP Server
 * Fix wrong indices in the eBPF Profiling related models.
+* Support exclude the specific namespaces traffic in the eBPF Access Log receiver.
 
 
 #### UI

diff --git a/...ore/src/main/java/org/apache/skywalking/oap/server/core/query/ProcessTopologyBuilder.java b/...ore/src/main/java/org/apache/skywalking/oap/server/core/query/ProcessTopologyBuilder.java
@@ -34,6 +34,7 @@
 import org.apache.skywalking.oap.server.core.storage.model.Model;
 import org.apache.skywalking.oap.server.core.storage.model.StorageModels;
 import org.apache.skywalking.oap.server.library.module.ModuleManager;
+import org.apache.skywalking.oap.server.library.util.CollectionUtils;
 
 import java.util.HashMap;
 import java.util.LinkedList;
@@ -71,6 +72,9 @@ ProcessTopology build(List<Call.CallDetail> clientCalls,
                           List<Call.CallDetail> serverCalls) throws Exception {
         log.debug("building process topology, total found client calls: {}, total found server calls: {}",
             clientCalls.size(), serverCalls.size());
+        if (CollectionUtils.isEmpty(clientCalls) && CollectionUtils.isEmpty(serverCalls)) {
+            return new ProcessTopology();
+        }
         List<Call> calls = new LinkedList<>();
         HashMap<String, Call> callMap = new HashMap<>();
 
@@ -88,7 +92,7 @@ ProcessTopology build(List<Call.CallDetail> clientCalls,
                     return p;
                 }).collect(Collectors.toList())).stream()
             .map(t -> (ProcessTraffic) t)
-            .collect(Collectors.toMap(m -> m.id().build(), this::buildNode));
+            .collect(Collectors.toMap(m -> m.id().build(), this::buildNode, (t1, t2) -> t1));
 
         int appendCallCount = 0;
         for (Call.CallDetail clientCall : clientCalls) {

diff --git a/.../apache/skywalking/oap/server/receiver/ebpf/provider/handler/AccessLogServiceHandler.java b/.../apache/skywalking/oap/server/receiver/ebpf/provider/handler/AccessLogServiceHandler.java
@@ -67,6 +67,7 @@
 import org.apache.skywalking.oap.server.telemetry.api.MetricsTag;
 
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -346,6 +347,7 @@ public static class NodeInfo {
         @Getter
         private final String clusterName;
         private final String nodeName;
+        private final List<String> excludeNamespaces;
 
         public NodeInfo(EBPFAccessLogNodeInfo node) {
             this.nodeName = node.getName();
@@ -354,6 +356,7 @@ public NodeInfo(EBPFAccessLogNodeInfo node) {
                     EBPFAccessLogNodeNetInterface::getIndex, EBPFAccessLogNodeNetInterface::getName, (a, b) -> a));
             this.bootTime = node.getBootTime();
             this.clusterName = node.getClusterName();
+            this.excludeNamespaces = buildExcludeNamespaces(node);
         }
 
         public String getNetInterfaceName(int index) {
@@ -365,10 +368,22 @@ public long parseMinuteTimeBucket(EBPFTimestamp timestamp) {
             return TimeBucket.getMinuteTimeBucket(seconds * 1000);
         }
 
+        public boolean shouldExcludeNamespace(String namespace) {
+            return excludeNamespaces.contains(namespace);
+        }
+
         public String toString() {
             return String.format("name: %s, clusterName: %s, network interfaces: %s",
                 nodeName, clusterName, netInterfaces);
         }
+
+        private List<String> buildExcludeNamespaces(EBPFAccessLogNodeInfo node) {
+            if (node.hasPolicy() && node.getPolicy().getExcludeNamespacesCount() > 0) {
+                return node.getPolicy().getExcludeNamespacesList().stream()
+                    .filter(StringUtil::isNotEmpty).collect(Collectors.toList());
+            }
+            return Collections.emptyList();
+        }
     }
 
     protected String buildServiceNameByAddress(NodeInfo nodeInfo, KubernetesProcessAddress address) {
@@ -451,6 +466,10 @@ protected KubernetesProcessAddress buildKubernetesAddressByIP(NodeInfo nodeInfo,
         if (pod == ObjectID.EMPTY) {
             return null;
         }
+        if (nodeInfo.shouldExcludeNamespace(pod.namespace())) {
+            log.debug("Should exclude the namespace[{}] traffic, pod: {}", pod.namespace(), pod.name());
+            return null;
+        }
         final ObjectID serviceName = K8sInfoRegistry.getInstance().findService(pod.namespace(), pod.name());
         if (serviceName == ObjectID.EMPTY) {
             return null;
@@ -661,9 +680,9 @@ private org.apache.skywalking.oap.server.core.source.DetectPoint parseToSourceRo
         }
 
         public String toString() {
-            return String.format("local: %s, remote: %s, role: %s, tlsMode: %s, protocolType: %s",
+            return String.format("local: %s, remote: %s, role: %s, tlsMode: %s, protocolType: %s, valid: %b",
                 buildConnectionAddressString(originalConnection.getLocal()),
-                buildConnectionAddressString(originalConnection.getRemote()), role, tlsMode, protocolType);
+                buildConnectionAddressString(originalConnection.getRemote()), role, tlsMode, protocolType, isValid());
         }
 
     }