(skywalking-rover) branch main updated: fix filter mechanism in the access log module (#148)

[li...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

liuhan pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/skywalking-rover.git


The following commit(s) were added to refs/heads/main by this push:
     new d39db69  fix filter mechanism in the access log module (#148)
d39db69 is described below

commit d39db696fc75944e01775d75a1d95e80879a5681
Author: mrproliu <74...@qq.com>
AuthorDate: Wed Oct 9 16:53:09 2024 +0700

    fix filter mechanism in the access log module (#148)
---
 pkg/accesslog/common/connection.go                 | 16 +++++++------
 pkg/accesslog/common/filter.go                     | 26 ++++++++++------------
 .../profiling/task/offcpu/c/profiling-analysis.yml |  2 +-
 3 files changed, 22 insertions(+), 22 deletions(-)

diff --git a/pkg/accesslog/common/connection.go b/pkg/accesslog/common/connection.go
index 690d781..4a960b7 100644
--- a/pkg/accesslog/common/connection.go
+++ b/pkg/accesslog/common/connection.go
@@ -482,7 +482,8 @@ func (c *ConnectionManager) isLocalTarget(pair *ip.SocketPair) addressProcessTyp
 
 func (c *ConnectionManager) AddNewProcess(pid int32, entities []api.ProcessInterface) {
 	// filtering the namespace
-	if c.shouldExcludeTheProcess(entities) {
+	monitorProcesses := c.shouldMonitorProcesses(entities)
+	if len(monitorProcesses) == 0 {
 		c.RemoveProcess(pid, entities)
 		return
 	}
@@ -491,9 +492,9 @@ func (c *ConnectionManager) AddNewProcess(pid int32, entities []api.ProcessInter
 	defer c.monitoringProcessLock.Unlock()
 
 	// adding monitoring process and IP addresses
-	c.monitoringProcesses[pid] = entities
+	c.monitoringProcesses[pid] = monitorProcesses
 	c.updateMonitorStatusForProcess(pid, true)
-	for _, entity := range entities {
+	for _, entity := range monitorProcesses {
 		for _, host := range entity.ExposeHosts() {
 			c.localIPWithPid[host] = pid
 		}
@@ -529,8 +530,8 @@ func (c *ConnectionManager) printTotalAddressesWithPid(prefix string) {
 	log.Debugf("----------------------------")
 }
 
-func (c *ConnectionManager) shouldExcludeTheProcess(entities []api.ProcessInterface) bool {
-	return c.monitorFilter.ShouldExclude(entities)
+func (c *ConnectionManager) shouldMonitorProcesses(entities []api.ProcessInterface) []api.ProcessInterface {
+	return c.monitorFilter.ShouldIncludeProcesses(entities)
 }
 
 func (c *ConnectionManager) RemoveProcess(pid int32, entities []api.ProcessInterface) {
@@ -549,10 +550,11 @@ func (c *ConnectionManager) RemoveProcess(pid int32, entities []api.ProcessInter
 func (c *ConnectionManager) RecheckAllProcesses(processes map[int32][]api.ProcessInterface) {
 	shouldMonitoringProcesses := make(map[int32][]api.ProcessInterface)
 	for pid, p := range processes {
-		if c.shouldExcludeTheProcess(p) {
+		monitorProcesses := c.shouldMonitorProcesses(p)
+		if len(monitorProcesses) == 0 {
 			continue
 		}
-		shouldMonitoringProcesses[pid] = p
+		shouldMonitoringProcesses[pid] = monitorProcesses
 	}
 	// checking the monitoring process
 	c.monitoringProcesses = shouldMonitoringProcesses
diff --git a/pkg/accesslog/common/filter.go b/pkg/accesslog/common/filter.go
index dad1741..0d60012 100644
--- a/pkg/accesslog/common/filter.go
+++ b/pkg/accesslog/common/filter.go
@@ -25,8 +25,8 @@ import (
 )
 
 type MonitorFilter interface {
-	// ShouldExclude returns true if the process should be excluded from monitoring.
-	ShouldExclude(process []api.ProcessInterface) bool
+	// ShouldIncludeProcesses returns true if the processes should be included in monitoring.
+	ShouldIncludeProcesses(process []api.ProcessInterface) []api.ProcessInterface
 	// ExcludeNamespaces returns a list of namespaces that should be excluded from monitoring.
 	ExcludeNamespaces() []string
 }
@@ -45,26 +45,24 @@ func NewStaticMonitorFilter(namespaces, clusters []string) *StaticMonitorFilter
 	}
 }
 
-func (s *StaticMonitorFilter) ShouldExclude(processes []api.ProcessInterface) bool {
-	containsNotExcludeCluster := false
+func (s *StaticMonitorFilter) ShouldIncludeProcesses(processes []api.ProcessInterface) (res []api.ProcessInterface) {
 	for _, entity := range processes {
 		if entity.DetectType() != api.Kubernetes { // for now, we only have the kubernetes detected processes
 			continue
 		}
 		namespace := entity.DetectProcess().(*kubernetes.Process).PodContainer().Pod.Namespace
-		if s.namespaces[namespace] {
-			return true
+		nameExclude := s.namespaces[namespace]
+		clusterExclude := false
+		if cluster, _, found := strings.Cut(entity.Entity().ServiceName, "::"); found && s.clusters[cluster] {
+			clusterExclude = true
 		}
-		if cluster, _, found := strings.Cut(entity.Entity().ServiceName, "::"); found {
-			if !s.clusters[cluster] {
-				containsNotExcludeCluster = true
-			}
-		} else {
-			containsNotExcludeCluster = true
-			break
+
+		// if the namespace and cluster are not excluded, include the process
+		if !nameExclude && !clusterExclude {
+			res = append(res, entity)
 		}
 	}
-	return !containsNotExcludeCluster
+	return
 }
 
 func (s *StaticMonitorFilter) ExcludeNamespaces() []string {
diff --git a/test/e2e/cases/profiling/task/offcpu/c/profiling-analysis.yml b/test/e2e/cases/profiling/task/offcpu/c/profiling-analysis.yml
index 3f03f03..8406052 100644
--- a/test/e2e/cases/profiling/task/offcpu/c/profiling-analysis.yml
+++ b/test/e2e/cases/profiling/task/offcpu/c/profiling-analysis.yml
@@ -20,7 +20,7 @@ trees:
   {{- contains .elements }}
   - id: "{{ notEmpty .id }}"
     parentid: "{{ notEmpty .parentid }}"
-    symbol: "fprintf"
+    symbol: "vfprintf"
     stacktype: USER_SPACE
     dumpcount: {{ gt .dumpcount 0 }}
 {{- end }}