Adding Access Log Service E2E Test (#127)

.github/workflows/rover.yaml

@@ -149,6 +149,10 @@ jobs:
             base: test/e2e/cases/process/istio
             config: e2e.yaml
             env: ISTIO_VERSION=1.13.1
+
+          - name: Access Log
+            base: test/e2e/cases/access_log
+            config: e2e.yaml
     steps:
       - uses: actions/checkout@v3
         with:

CHANGES.md

@@ -8,6 +8,7 @@ Release Notes.
 * Upgrade LLVM to `18`.
 
 #### Bug Fixes
+* Fixed the issue where `conntrack` could not find the Reply IP in the access log module.
 
 #### Documentation
 

pkg/accesslog/collector/connect.go

@@ -19,6 +19,7 @@ package collector
 
 import (
 	"encoding/binary"
+	"net"
 
 	"github.com/sirupsen/logrus"
 
@@ -99,6 +100,26 @@ func (c *ConnectCollector) Start(_ *module.Manager, context *common.AccessLogCon
 func (c *ConnectCollector) Stop() {
 }
 
+func (c *ConnectCollector) fixSocketFamilyIfNeed(event *events.SocketConnectEvent, result *ip.SocketPair) {
+	if result == nil {
+		return
+	}
+	if parseIP := net.ParseIP(result.SrcIP); parseIP != nil {
+		var actual uint32
+		if parseIP.To4() != nil {
+			actual = unix.AF_INET
+		} else {
+			actual = unix.AF_INET6
+		}
+
+		if result.Family != actual {
+			connectLogger.Debugf("fix the socket family from %d to %d, connection ID: %d, randomID: %d",
+				result.Family, actual, event.ConID, event.RandomID)
+			result.Family = actual
+		}
+	}
+}
+
 func (c *ConnectCollector) buildSocketFromConnectEvent(event *events.SocketConnectEvent) *ip.SocketPair {
 	if event.SocketFamily != unix.AF_INET && event.SocketFamily != unix.AF_INET6 && event.SocketFamily != enums.SocketFamilyUnknown {
 		// if not ipv4, ipv6 or unknown, ignore
@@ -122,6 +143,7 @@ func (c *ConnectCollector) buildSocketFromConnectEvent(event *events.SocketConne
 	connectLogger.Debugf("found the connection from the socket, connection ID: %d, randomID: %d",
 		event.ConID, event.RandomID)
 	pair.Role = enums.ConnectionRole(event.Role)
+	c.fixSocketFamilyIfNeed(event, pair)
 	c.tryToUpdateSocketFromConntrack(event, pair)
 	return pair
 }
@@ -193,12 +215,14 @@ func (c *ConnectCollector) buildSocketPair(event *events.SocketConnectEvent) *ip
 		return result
 	}
 
+	c.fixSocketFamilyIfNeed(event, result)
 	c.tryToUpdateSocketFromConntrack(event, result)
 	return result
 }
 
 func (c *ConnectCollector) tryToUpdateSocketFromConntrack(event *events.SocketConnectEvent, socket *ip.SocketPair) {
-	if socket != nil && socket.IsValid() && c.connTracker != nil && !tools.IsLocalHostAddress(socket.DestIP) {
+	if socket != nil && socket.IsValid() && c.connTracker != nil && !tools.IsLocalHostAddress(socket.DestIP) &&
+		event.FuncName != enums.SocketFunctionNameAccept { // accept event don't need to update the remote address
 		// if no contract and socket data is valid, then trying to get the remote address from the socket
 		// to encase the remote address is not the real remote address
 		originalIP := socket.DestIP

pkg/accesslog/runner.go

@@ -241,8 +241,9 @@ func (r *Runner) buildAccessLogMessage(firstLog, firstConnection bool, conn *com
 		rpcCon = conn.RPCConnection
 		if log.Enable(logrus.DebugLevel) {
 			log.Debugf("ready to sending access log with connection, connection ID: %d, random ID: %d, "+
-				"local: %s, remote: %s, role: %s",
-				conn.ConnectionID, conn.RandomID, rpcCon.Local, rpcCon.Remote, rpcCon.Role)
+				"local: %s, remote: %s, role: %s, kernel logs count: %d, contains protocol log: %t",
+				conn.ConnectionID, conn.RandomID, rpcCon.Local, rpcCon.Remote, rpcCon.Role,
+				len(kernelLogs), protocolLog != nil)
 		}
 	}
 	return &v3.EBPFAccessLogMessage{

pkg/tools/ip/conntrack.go

@@ -30,7 +30,16 @@ import (
 
 var log = logger.GetLogger("tools", "ip")
 
-var numberStrategies = map[string]uint8{"tcp": syscall.IPPROTO_TCP, "udp": syscall.IPPROTO_UDP}
+var numberStrategies = []struct {
+	name  string
+	proto uint8
+}{{
+	name:  "tcp",
+	proto: syscall.IPPROTO_TCP,
+}, {
+	name:  "udp",
+	proto: syscall.IPPROTO_UDP,
+}}
 
 type ConnTrack struct {
 	tracker *conntrack.Nfct
@@ -51,13 +60,15 @@ func (c *ConnTrack) UpdateRealPeerAddress(addr *SocketPair) bool {
 	}
 
 	tuple := c.parseSocketToTuple(addr)
-	for name, strategy := range numberStrategies {
-		tuple.Proto.Number = &strategy
+	for _, info := range numberStrategies {
+		tuple.Proto.Number = &(info.proto)
 
+		// using get to query protocol
 		session, e := c.tracker.Get(conntrack.Conntrack, family, conntrack.Con{Origin: tuple})
 		if e != nil {
-			// try to get the reply session, if the strategy not exists or from accept events, have error is normal
-			log.Debugf("cannot get the conntrack session, strategy: %s, error: %v", name, e)
+			// try to get the reply session, if the info not exists or from accept events, have error is normal
+			log.Debugf("cannot get the conntrack session, type: %s, family: %d, origin src: %s:%d, origin dest: %s:%d, error: %v", info.name,
+				family, tuple.Src, *tuple.Proto.SrcPort, tuple.Dst, *tuple.Proto.DstPort, e)
 			continue
 		}
 
@@ -67,6 +78,16 @@ func (c *ConnTrack) UpdateRealPeerAddress(addr *SocketPair) bool {
 		}
 	}
 
+	// using dump to query protocol
+	dump, e := c.tracker.Dump(conntrack.Conntrack, family)
+	if e != nil {
+		log.Debug("cannot dump the conntrack session, error: ", e)
+		return false
+	}
+	if res := c.filterValidateReply(dump, tuple); res != nil {
+		addr.DestIP = res.Src.String()
+		return true
+	}
 	return false
 }
 

test/e2e/base/env

@@ -13,8 +13,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-SW_CTL_COMMIT=6b2eb0011e38b630db6af7203db215806bd141ed
-SW_OAP_COMMIT=9ba0ad299139eeb4bb4e274c500cec1deaf84f79
+SW_CTL_COMMIT=ee371a210afe2dc6e65c2229b6a0519f8a4a2752
+SW_OAP_COMMIT=7a8bbacd20381bc780c48abf5706fd6e529c9872
 SW_KUBERNETES_COMMIT_SHA=a14f386063fffc61ed9b396e1328b76d33239aba
 
 SW_AGENT_GO_COMMIT=216f122d942cb683f48578d3014cc5ea83637582