Delayed dissection changes

go.mod

@@ -20,7 +20,7 @@ require (
 	github.com/jinzhu/copier v0.4.0
 	github.com/knightsc/gapstone v0.0.0-20191231144527-6fa5afaf11a9
 	github.com/kubeshark/api v1.2.2-0.20250617143439-101885933817
-	github.com/kubeshark/api2 v0.0.0-20250925155137-526ac2c1c2da
+	github.com/kubeshark/api2 v0.0.0-20251028135108-ac52a62137ab
 	github.com/kubeshark/gopacket v1.1.43
 	github.com/kubeshark/procfs v0.0.0-20250312150455-4b9efb18c324
 	github.com/kubeshark/tracerproto v1.0.2

go.sum

@@ -124,8 +124,8 @@ github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
 github.com/kubeshark/api v1.2.2-0.20250617143439-101885933817 h1:IdlCmIBpfomskM3oBUm9cJiRDPKw0Skur4+kXHi3HIs=
 github.com/kubeshark/api v1.2.2-0.20250617143439-101885933817/go.mod h1:jTq+eKUczqXIS1Ru7MD0u507kCzW/uTrhaEHeKLTCoI=
-github.com/kubeshark/api2 v0.0.0-20250925155137-526ac2c1c2da h1:aLpVyeqXa0Tg/net6sxzTjdD5Yu9ac4RDkJrEeHsr5Y=
-github.com/kubeshark/api2 v0.0.0-20250925155137-526ac2c1c2da/go.mod h1:U7Y90izLanlHFPnljwcnAtDpyiFGrrF2bp67w1GXA5U=
+github.com/kubeshark/api2 v0.0.0-20251028135108-ac52a62137ab h1:Z8R9IXikuhWSS0nLH+mcG/td4g8n9dLUyOiFdtQ317U=
+github.com/kubeshark/api2 v0.0.0-20251028135108-ac52a62137ab/go.mod h1:u+RE/jKLCmogSqEVuTgV3/g5ae2g2Jr/Bz5L/w8rLZc=
 github.com/kubeshark/gopacket v1.1.43 h1:zlsejFqXFG2wiG7noraCRrVgqjjIA1lnbfdMbnyDouw=
 github.com/kubeshark/gopacket v1.1.43/go.mod h1:Qo8/i/tdT74CCT7/pjO0L55Pktv5dQfj7M/Arv8MKm8=
 github.com/kubeshark/procfs v0.0.0-20250312150455-4b9efb18c324 h1:bkkaNmy70yebjPIPeZv5bDamFl0fmAKQn038NWf/GBU=

pkg/cgroup/cgoup.go

@@ -60,7 +60,7 @@ type Cgroups struct {
 	hid      int     // default cgroup controller hierarchy ID
 }
 
-func NewCgroups() (*Cgroups, error) {
+func NewCgroups(procfs string) (*Cgroups, error) {
 	var err error
 	var cgrp *Cgroup
 	var cgroupv1, cgroupv2 Cgroup
@@ -73,7 +73,7 @@ func NewCgroups() (*Cgroups, error) {
 
 	// only start cgroupv1 if it is the OS default (or else it isn't needed)
 	if defaultVersion == CgroupVersion1 {
-		cgroupv1, err = NewCgroup(CgroupVersion1)
+		cgroupv1, err = NewCgroup(procfs, CgroupVersion1)
 		if err != nil {
 			if _, ok := err.(*VersionNotSupported); !ok {
 				return nil, errfmt.WrapError(err)
@@ -82,7 +82,7 @@ func NewCgroups() (*Cgroups, error) {
 	}
 
 	// start cgroupv2 (if supported)
-	cgroupv2, err = NewCgroup(CgroupVersion2)
+	cgroupv2, err = NewCgroup(procfs, CgroupVersion2)
 	if err != nil {
 		if _, ok := err.(*VersionNotSupported); !ok {
 			return nil, errfmt.WrapError(err)
@@ -172,14 +172,14 @@ type Cgroup interface {
 	GetVersion() CgroupVersion
 }
 
-func NewCgroup(ver CgroupVersion) (Cgroup, error) {
+func NewCgroup(procfs string, ver CgroupVersion) (Cgroup, error) {
 	var c Cgroup
 
 	switch ver {
 	case CgroupVersion1:
-		c = &CgroupV1{}
+		c = &CgroupV1{procfs: procfs}
 	case CgroupVersion2:
-		c = &CgroupV2{}
+		c = &CgroupV2{procfs: procfs}
 	}
 
 	return c, c.init()
@@ -188,14 +188,15 @@ func NewCgroup(ver CgroupVersion) (Cgroup, error) {
 // cgroupv1
 
 type CgroupV1 struct {
+	procfs     string
 	mounted    *mount.MountHostOnce
 	mountpoint string
 	hid        int
 }
 
 func (c *CgroupV1) init() error {
 	// 0. check if cgroup type is supported
-	supported, err := mount.IsFileSystemSupported(CgroupVersion1.String())
+	supported, err := mount.IsFileSystemSupported(c.procfs, CgroupVersion1.String())
 	if err != nil {
 		return errfmt.WrapError(err)
 	}
@@ -205,6 +206,7 @@ func (c *CgroupV1) init() error {
 
 	// 1. mount cgroup (if needed)
 	c.mounted, err = mount.NewMountHostOnce(
+		c.procfs,
 		CgroupV1FsType,
 		CgroupV1FsType,
 		CgroupDefaultController,
@@ -239,14 +241,15 @@ func (c *CgroupV1) GetVersion() CgroupVersion {
 // cgroupv2
 
 type CgroupV2 struct {
+	procfs     string
 	mounted    *mount.MountHostOnce
 	mountpoint string
 	hid        int
 }
 
 func (c *CgroupV2) init() error {
 	// 0. check if cgroup type is supported
-	supported, err := mount.IsFileSystemSupported(CgroupVersion2.String())
+	supported, err := mount.IsFileSystemSupported(c.procfs, CgroupVersion2.String())
 	if err != nil {
 		return errfmt.WrapError(err)
 	}
@@ -256,6 +259,7 @@ func (c *CgroupV2) init() error {
 
 	// 1. mount cgroup (if needed)
 	c.mounted, err = mount.NewMountHostOnce(
+		c.procfs,
 		CgroupV2FsType,
 		CgroupV2FsType,
 		"",          // cgroupv2 has no default controller

pkg/cgroup/controller.go

@@ -276,7 +276,7 @@ func NewCgroupsController(procfs string, grpcServer *grpcservice.GRPCService) (C
 		actualCgroupVersion = CgroupVersion2
 	}
 	cgroupSupported := true
-	cgroupV2, err := NewCgroup(CgroupVersion2)
+	cgroupV2, err := NewCgroup(procfs, CgroupVersion2)
 	if err != nil {
 		if _, ok := err.(*VersionNotSupported); !ok {
 			return nil, fmt.Errorf("new cgroup2 create failed")

pkg/mount/mount.go

@@ -19,8 +19,8 @@ import (
 // Constants
 
 const (
-	procMounts      = "/hostproc/self/mountinfo"
-	procFilesystems = "/hostproc/filesystems"
+	procMounts      = "self/mountinfo"
+	procFilesystems = "filesystems"
 	tmpPathPrefix   = "kubeshark"
 )
 
@@ -34,6 +34,7 @@ const (
 // and manage it (umounting at its destruction). If already mounted, the filesystem
 // is left untouched at object's destruction.
 type MountHostOnce struct {
+	procfs  string
 	source  string
 	target  string
 	fsType  string
@@ -42,8 +43,9 @@ type MountHostOnce struct {
 	mounted bool
 }
 
-func NewMountHostOnce(source, fstype, data, where string) (*MountHostOnce, error) {
+func NewMountHostOnce(procfs, source, fstype, data, where string) (*MountHostOnce, error) {
 	m := &MountHostOnce{
+		procfs: procfs,
 		source: source, // device and/or pseudo-filesystem to mount
 		fsType: fstype, // fs type
 		data:   data,   // extra data
@@ -151,7 +153,7 @@ func (m *MountHostOnce) GetMountpoint() string {
 func (m *MountHostOnce) isMountedByOS(where string) (bool, error) {
 	var err error
 	var mp string
-	mp, err = SearchMountpointFromHost(m.fsType, m.data)
+	mp, err = SearchMountpointFromHost(m.procfs, m.fsType, m.data)
 	if err != nil {
 		return false, errfmt.WrapError(err)
 	}
@@ -175,10 +177,10 @@ func (m *MountHostOnce) isMountedByOS(where string) (bool, error) {
 //
 
 // IsFileSystemSupported checks if given fs is supported by the running kernel
-func IsFileSystemSupported(fsType string) (bool, error) {
-	file, err := os.Open(procFilesystems)
+func IsFileSystemSupported(procfs, fsType string) (bool, error) {
+	file, err := os.Open(filepath.Join(procfs, procFilesystems))
 	if err != nil {
-		return false, CouldNotOpenFile(procFilesystems, err)
+		return false, CouldNotOpenFile(filepath.Join(procfs, procFilesystems), err)
 	}
 	defer func() {
 		if err := file.Close(); err != nil {
@@ -201,10 +203,10 @@ func IsFileSystemSupported(fsType string) (bool, error) {
 // SearchMountpointFromHost returns the last mountpoint for a given filesystem type
 // containing a searchable string. It confirms the mount originates from the root file
 // system.
-func SearchMountpointFromHost(fstype string, search string) (string, error) {
+func SearchMountpointFromHost(procfs, fstype string, search string) (string, error) {
 	mp := ""
 
-	file, err := os.Open(procMounts)
+	file, err := os.Open(filepath.Join(procfs, procMounts))
 	if err != nil {
 		return "", errfmt.WrapError(err)
 	}

pkg/poller/packets/packets_poller.go

@@ -246,11 +246,11 @@ func (p *PacketsPoller) logPeriodicStats() {
 		packetsPerSec := float64(packetsDelta) / elapsed
 		bytesPerSec := float64(bytesDelta) / elapsed
 
-		log.Info().
+		log.Debug().
 			Float64("chunks_per_sec", chunksPerSec).
 			Float64("packets_per_sec", packetsPerSec).
 			Str("bytes_per_sec", formatBytes(uint64(bytesPerSec))).
-			Msg("PacketsPoller statistics")
+			Msg("PacketsPoller stats")
 
 		// Update last stats and time
 		p.lastStats = p.stats

pkg/poller/poller.go

@@ -20,11 +20,11 @@ type BpfPollerImpl struct {
 	logPoller     *logPoller.BpfLogger
 }
 
-func NewBpfPoller(bpfObjs *bpf.BpfObjects, cgroupsController cgroup.CgroupsController, systemStoreManager *rawcapture.Manager, tlsLogDisabled bool) (BpfPoller, error) {
+func NewBpfPoller(procfs string, bpfObjs *bpf.BpfObjects, cgroupsController cgroup.CgroupsController, systemStoreManager *rawcapture.Manager, tlsLogDisabled bool) (BpfPoller, error) {
 	var err error
 	p := BpfPollerImpl{}
 
-	if p.syscallPoller, err = syscallPoller.NewSyscallEventsTracer(bpfObjs, cgroupsController, systemStoreManager); err != nil {
+	if p.syscallPoller, err = syscallPoller.NewSyscallEventsTracer(procfs, bpfObjs, cgroupsController, systemStoreManager); err != nil {
 		return nil, fmt.Errorf("create syscall poller failed: %v", err)
 	}
 

pkg/poller/syscall/tracer_syscall_events.go

@@ -25,19 +25,21 @@ import (
 )
 
 type SyscallEventsTracer struct {
+	procfs             string
 	cgroupController   cgroup.CgroupsController
 	eventReader        *perf.Reader
 	eventSocket        *socket.SocketEvent
 	systemStoreManager *rawcapture.Manager
 }
 
-func NewSyscallEventsTracer(bpfObjs *bpf.BpfObjects, cgroupController cgroup.CgroupsController, systemStoreManager *rawcapture.Manager) (*SyscallEventsTracer, error) {
+func NewSyscallEventsTracer(procfs string, bpfObjs *bpf.BpfObjects, cgroupController cgroup.CgroupsController, systemStoreManager *rawcapture.Manager) (*SyscallEventsTracer, error) {
 	reader, err := perf.NewReader(bpfObjs.BpfObjs.SyscallEvents, os.Getpagesize())
 	if err != nil {
 		return nil, fmt.Errorf("open events perf buffer failed")
 	}
 
 	return &SyscallEventsTracer{
+		procfs:             procfs,
 		cgroupController:   cgroupController,
 		eventReader:        reader,
 		eventSocket:        socket.NewSocketEvent(misc.GetSyscallEventSocketPath()),
@@ -105,7 +107,7 @@ func (t *SyscallEventsTracer) pollEvents() {
 		var e events.SyscallEvent
 		e.SyscallEventMessage = *ev
 
-		e.ProcessPath, _ = resolver.ResolveSymlinkWithoutValidation(filepath.Join("/hostproc", fmt.Sprintf("%v", ev.HostPid), "exe"))
+		e.ProcessPath, _ = resolver.ResolveSymlinkWithoutValidation(filepath.Join(t.procfs, fmt.Sprintf("%v", ev.HostPid), "exe"))
 		log.Debug().Msg(fmt.Sprintf("Syscall event %v: %v:%v->%v:%v command: %v host pid: %v host ppid: %v pid: %v ppid: %v cgroup id: %v, sent (pkts: %v, bytes: %v), recv (pkts: %v, bytes: %v)",
 			evName,
 			toIP(e.IpSrc),
@@ -132,15 +134,16 @@ func (t *SyscallEventsTracer) pollEvents() {
 			EventId:       uint32(e.EventId),
 			IpSrc:         &commonv1.IP{Ip: ipv4ToIPv6Mapped(e.IpSrc)},
 			IpDst:         &commonv1.IP{Ip: ipv4ToIPv6Mapped(e.IpDst)},
-			PortSrc:       uint32(e.PortSrc),
-			PortDst:       uint32(e.PortDst),
+			PortSrc:       uint32(toPort(e.PortSrc)),
+			PortDst:       uint32(toPort(e.PortDst)),
 			CgroupId:      e.CgroupID,
 			HostPid:       uint32(e.HostPid),
 			HostParentPid: uint32(e.HostParentPid),
 			Pid:           uint32(e.Pid),
 			ParentPid:     uint32(e.ParentPid),
 			Command:       e.CmdPath(),
 			ProcessPath:   e.ProcessPath,
+			ContainerId:   t.cgroupController.GetContainerID(e.CgroupID),
 		}
 		t.systemStoreManager.EnqueueSyscall(bin)
 	}
@@ -150,6 +153,6 @@ func ipv4ToIPv6Mapped(v uint32) []byte {
 	b := make([]byte, 16)
 	b[10] = 0xff
 	b[11] = 0xff
-	binary.BigEndian.PutUint32(b[12:], v)
+	binary.LittleEndian.PutUint32(b[12:], v)
 	return b
 }

pkg/resolver/connections.go

@@ -34,7 +34,7 @@ func GatherPidsTCPMap(procfs string, isCgroupV2 bool) (tcpMap map[uint64]bool, e
 
 		var conns []IpSocketLine
 		for i := range pids {
-			conns, err = getTCPConnections(fmt.Sprintf("%v", pids[i].hostPid))
+			conns, err = getTCPConnections(procfs, fmt.Sprintf("%v", pids[i].hostPid))
 			if err != nil {
 				// process can be short-living
 				continue

pkg/resolver/resolver.go

@@ -154,9 +154,9 @@ func getAllFlows(procfs string, isCgroupV2 bool, proto string, addFlowEntry addF
 		var conns []IpSocketLine
 		for i := range pids {
 			if proto == "tcp" {
-				conns, err = getTCPConnections(fmt.Sprintf("%v", pids[i].hostPid))
+				conns, err = getTCPConnections(procfs, fmt.Sprintf("%v", pids[i].hostPid))
 			} else {
-				conns, err = getUDPConnections(fmt.Sprintf("%v", pids[i].hostPid))
+				conns, err = getUDPConnections(procfs, fmt.Sprintf("%v", pids[i].hostPid))
 			}
 			if err != nil {
 				// process can be short-living
@@ -240,6 +240,9 @@ func getAllFlows(procfs string, isCgroupV2 bool, proto string, addFlowEntry addF
 }
 
 func getAllCgroups(procfs string, isCgroupV2 bool) (ret map[string]uint64, err error) {
+	if procfs == "" {
+		return nil, nil
+	}
 	ret = make(map[string]uint64)
 	procDir, err := os.Open(procfs)
 	if err != nil {
@@ -491,10 +494,10 @@ func findCgroupsEndWith(isCgroupsV2 bool, pidPaths map[string]struct{}) (paths m
 	return paths, err
 }
 
-func getTCPConnections(pid string) (lines []IpSocketLine, err error) {
-	return getSocketLines("tcp", pid)
+func getTCPConnections(procfs string, pid string) (lines []IpSocketLine, err error) {
+	return getSocketLines(procfs, "tcp", pid)
 }
 
-func getUDPConnections(pid string) (lines []IpSocketLine, err error) {
-	return getSocketLines("udp", pid)
+func getUDPConnections(procfs string, pid string) (lines []IpSocketLine, err error) {
+	return getSocketLines(procfs, "udp", pid)
 }

pkg/resolver/utils.go

@@ -27,16 +27,16 @@ type IpSocketLine struct {
 	Inode     uint64
 }
 
-func getSocketLines(proto, pid string) (lines []IpSocketLine, err error) {
+func getSocketLines(procpath, proto, pid string) (lines []IpSocketLine, err error) {
 	var tcpConns procfs.NetTCP
 	var tcpConns6 procfs.NetTCP
-	getTcpConns := func(_pid string) error {
-		tcpConns, err = procfs.NewNetTCP(fmt.Sprintf("/hostproc/%v/net/tcp", _pid))
+	getTcpConns := func(procpath, _pid string) error {
+		tcpConns, err = procfs.NewNetTCP(fmt.Sprintf("%v/%v/net/tcp", procpath, _pid))
 		if err != nil {
 			return err
 		}
 
-		tcpConns6, err = procfs.NewNetTCP(fmt.Sprintf("/hostproc/%v/net/tcp6", _pid))
+		tcpConns6, err = procfs.NewNetTCP(fmt.Sprintf("%v/%v/net/tcp6", procpath, _pid))
 		if errors.Is(err, os.ErrNotExist) {
 			// ipv6 is disabled
 			return nil
@@ -49,13 +49,13 @@ func getSocketLines(proto, pid string) (lines []IpSocketLine, err error) {
 
 	var udpConns procfs.NetUDP
 	var udpConns6 procfs.NetUDP
-	getUdpConns := func(_pid string) error {
-		udpConns, err = procfs.NewNetUDP(fmt.Sprintf("/hostproc/%v/net/udp", _pid))
+	getUdpConns := func(procpath, _pid string) error {
+		udpConns, err = procfs.NewNetUDP(fmt.Sprintf("%v/%v/net/udp", procpath, _pid))
 		if err != nil {
 			return err
 		}
 
-		udpConns6, err = procfs.NewNetUDP(fmt.Sprintf("/hostproc/%v/net/udp6", _pid))
+		udpConns6, err = procfs.NewNetUDP(fmt.Sprintf("%v/%v/net/udp6", procpath, _pid))
 		if errors.Is(err, os.ErrNotExist) {
 			// ipv6 is disabled
 			return nil
@@ -67,7 +67,7 @@ func getSocketLines(proto, pid string) (lines []IpSocketLine, err error) {
 	}
 
 	if proto == "tcp" {
-		if err = getTcpConns(pid); err != nil {
+		if err = getTcpConns(procpath, pid); err != nil {
 			err = fmt.Errorf("execute tcp in ns failed for pid: %v error: %v", pid, err)
 			return lines, err
 		}
@@ -106,7 +106,7 @@ func getSocketLines(proto, pid string) (lines []IpSocketLine, err error) {
 		}
 
 	} else if proto == "udp" {
-		if err = getUdpConns(pid); err != nil {
+		if err = getUdpConns(procpath, pid); err != nil {
 			err = fmt.Errorf("execute udp in ns failed for pid: %v error: %v", pid, err)
 			return lines, err
 		}

tracer.go

@@ -121,7 +121,7 @@ func (t *Tracer) Init(
 		}
 	}
 
-	allPollers, err := poller.NewBpfPoller(t.bpfObjects, t.cgroupsController, systemStoreManager, *disableTlsLog)
+	allPollers, err := poller.NewBpfPoller(procfs, t.bpfObjects, t.cgroupsController, systemStoreManager, *disableTlsLog)
 	if err != nil {
 		return fmt.Errorf("create eBPF poler failed failed: %v", err)
 	}