chore: merge upstream changes

cgroup/cgroup.go

@@ -170,7 +170,7 @@ func containerByCgroup(path string) (ContainerType, string, error) {
 		if matches == nil {
 			return ContainerTypeUnknown, "", fmt.Errorf("invalid systemd cgroup %s", path)
 		}
-		return ContainerTypeSystemdService, matches[1], nil
+		return ContainerTypeSystemdService, strings.Replace(matches[1], "\\x2d", "-", -1), nil
 	}
 	return ContainerTypeUnknown, "", fmt.Errorf("unknown container: %s", path)
 }

cgroup/cgroup_test.go

@@ -119,7 +119,7 @@ func TestContainerByCgroup(t *testing.T) {
 
 	typ, id, err = containerByCgroup("/system.slice/system-serial\\x2dgetty.slice")
 	as.Equal(typ, ContainerTypeSystemdService)
-	as.Equal("/system.slice/system-serial\\x2dgetty.slice", id)
+	as.Equal("/system.slice/system-serial-getty.slice", id)
 	as.Nil(err)
 
 	typ, id, err = containerByCgroup("/runtime.slice/kubelet.service")

cgroup/cpu.go

@@ -2,7 +2,7 @@ package cgroup
 
 import (
 	"fmt"
-	"io/ioutil"
+	"os"
 	"path"
 	"strconv"
 	"strings"
@@ -57,28 +57,26 @@ func (cg Cgroup) cpuStatV2() (*CPUStat, error) {
 		UsageSeconds:         float64(vars["usage_usec"]) / 1e6,
 		ThrottledTimeSeconds: float64(vars["throttled_usec"]) / 1e6,
 	}
-	payload, err := ioutil.ReadFile(path.Join(cgRoot, cg.subsystems[""], "cpu.max"))
-	if err != nil {
-		return nil, err
-	}
-	data := strings.TrimSpace(string(payload))
-	parts := strings.Fields(data)
-	if len(parts) != 2 {
-		return nil, fmt.Errorf("invalid cpu.max payload: %s", data)
-	}
-	if parts[0] == "max" { //no limit
-		return res, nil
-	}
-	quotaUs, err := strconv.ParseUint(parts[0], 10, 64)
-	if err != nil {
-		return nil, fmt.Errorf("invalid quota value in cpu.max: %s", parts[0])
-	}
-	periodUs, err := strconv.ParseUint(parts[1], 10, 64)
-	if err != nil {
-		return nil, fmt.Errorf("invalid period value in cpu.max: %s", parts[1])
-	}
-	if periodUs > 0 {
-		res.LimitCores = float64(quotaUs) / float64(periodUs)
+	if payload, err := os.ReadFile(path.Join(cgRoot, cg.subsystems[""], "cpu.max")); err == nil {
+		data := strings.TrimSpace(string(payload))
+		parts := strings.Fields(data)
+		if len(parts) != 2 {
+			return nil, fmt.Errorf("invalid cpu.max payload: %s", data)
+		}
+		if parts[0] == "max" { //no limit
+			return res, nil
+		}
+		quotaUs, err := strconv.ParseUint(parts[0], 10, 64)
+		if err != nil {
+			return nil, fmt.Errorf("invalid quota value in cpu.max: %s", parts[0])
+		}
+		periodUs, err := strconv.ParseUint(parts[1], 10, 64)
+		if err != nil {
+			return nil, fmt.Errorf("invalid period value in cpu.max: %s", parts[1])
+		}
+		if periodUs > 0 {
+			res.LimitCores = float64(quotaUs) / float64(periodUs)
+		}
 	}
 	return res, nil
 }

cgroup/utils.go

@@ -1,15 +1,15 @@
 package cgroup
 
 import (
-	"io/ioutil"
+	"os"
 	"strconv"
 	"strings"
 
 	"k8s.io/klog/v2"
 )
 
 func readVariablesFromFile(filePath string) (map[string]uint64, error) {
-	data, err := ioutil.ReadFile(filePath)
+	data, err := os.ReadFile(filePath)
 	if err != nil {
 		return nil, err
 	}
@@ -29,15 +29,15 @@ func readVariablesFromFile(filePath string) (map[string]uint64, error) {
 }
 
 func readIntFromFile(filePath string) (int64, error) {
-	data, err := ioutil.ReadFile(filePath)
+	data, err := os.ReadFile(filePath)
 	if err != nil {
 		return 0, err
 	}
 	return strconv.ParseInt(strings.TrimSpace(string(data)), 10, 64)
 }
 
 func readUintFromFile(filePath string) (uint64, error) {
-	data, err := ioutil.ReadFile(filePath)
+	data, err := os.ReadFile(filePath)
 	if err != nil {
 		return 0, err
 	}

containers/container.go

@@ -130,7 +130,6 @@ type Container struct {
 	delaysLock  sync.Mutex
 
 	listens map[netaddr.IPPort]map[uint32]*ListenDetails
-	ipsByNs map[string][]netaddr.IP
 
 	connectsSuccessful map[AddrPair]*ConnectionStats // dst:actual_dst -> count
 	connectsFailed     map[netaddr.IPPort]int64      // dst -> count
@@ -177,7 +176,6 @@ func NewContainer(id ContainerID, cg *cgroup.Cgroup, md *ContainerMetadata, host
 		delaysByPid: map[uint32]Delays{},
 
 		listens: map[netaddr.IPPort]map[uint32]*ListenDetails{},
-		ipsByNs: map[string][]netaddr.IP{},
 
 		connectsSuccessful: map[AddrPair]*ConnectionStats{},
 		connectsFailed:     map[netaddr.IPPort]int64{},
@@ -525,16 +523,12 @@ func (c *Container) onListenOpen(pid uint32, addr netaddr.IPPort, safe bool) {
 			return
 		}
 		defer ns.Close()
-		nsId := ns.UniqueId()
-		ips, ok := c.ipsByNs[nsId]
-		if !ok {
-			if ips, err = proc.GetNsIps(ns); err != nil {
-				klog.Warningln(err)
-			} else {
-				klog.Infof("got IPs %s for %s", ips, nsId)
-				c.ipsByNs[nsId] = ips
-			}
+		ips, err := proc.GetNsIps(ns)
+		if err != nil {
+			klog.Warningln(err)
+			return
 		}
+		klog.Infof("got IPs %s for %s", ips, ns.UniqueId())
 		details.NsIPs = ips
 	}
 }
@@ -662,31 +656,20 @@ func (c *Container) getActualDestination(p *Process, src, dst netaddr.IPPort) (*
 	return nil, nil
 }
 
-func (c *Container) onConnectionClose(e ebpftracer.Event) bool {
-	srcDst := AddrPair{src: e.SrcAddr, dst: e.DstAddr}
+func (c *Container) onConnectionClose(e ebpftracer.Event) {
 	c.lock.Lock()
-	conn, ok := c.connectionsActive[srcDst]
+	conn := c.connectionsByPidFd[PidFd{Pid: e.Pid, Fd: e.Fd}]
 	c.lock.Unlock()
 	if conn != nil {
 		if conn.Closed.IsZero() {
-			if e.Pid == 0 && e.Fd == 0 {
-				stats, err := c.registry.tracer.GetAndDeleteTCPConnection(conn.Pid, conn.Fd)
-				if err != nil {
-					klog.Warningln(c.id, conn.Pid, conn.Fd, conn.ActualDest, err)
-				} else {
-					c.lock.Lock()
-					c.updateConnectionTrafficStats(conn, stats.BytesSent, stats.BytesReceived)
-					c.lock.Unlock()
-				}
-			} else if e.TrafficStats != nil {
+			if e.TrafficStats != nil {
 				c.lock.Lock()
 				c.updateConnectionTrafficStats(conn, e.TrafficStats.BytesSent, e.TrafficStats.BytesReceived)
 				c.lock.Unlock()
 			}
 			conn.Closed = time.Now()
 		}
 	}
-	return ok
 }
 
 func (c *Container) updateTrafficStats(u *TrafficStatsUpdate) {
@@ -1148,12 +1131,6 @@ func (c *Container) gc(now time.Time) {
 		seenNamespaces[p.NetNsId()] = true
 	}
 
-	for ns := range c.ipsByNs {
-		if !seenNamespaces[ns] {
-			delete(c.ipsByNs, ns)
-		}
-	}
-
 	c.revalidateListens(now, listens)
 
 	for srcDst, conn := range c.connectionsActive {

containers/crio.go

@@ -8,6 +8,7 @@ import (
 	"net"
 	"net/http"
 	"os"
+	"strings"
 	"time"
 
 	"github.com/coroot/coroot-node-agent/common"
@@ -20,7 +21,6 @@ const crioTimeout = 30 * time.Second
 
 var (
 	crioClient *http.Client
-	crioSocket = proc.HostPath("/var/run/crio/crio.sock")
 )
 
 type CrioContainerInfo struct {
@@ -37,8 +37,23 @@ type CrioVolume struct {
 }
 
 func CrioInit() error {
-	if _, err := os.Stat(crioSocket); err != nil {
-		return err
+	sockets := []string{
+		"/var/run/crio/crio.sock",
+		"/run/crio/crio.sock",
+	}
+	var crioSocket string
+	var err error
+	for _, socket := range sockets {
+		socketHostPath := proc.HostPath(socket)
+		if _, err := os.Stat(socketHostPath); err == nil {
+			crioSocket = socketHostPath
+			break
+		}
+	}
+	if err != nil {
+		return fmt.Errorf("couldn't connect to CRI-O through the following UNIX sockets: [%s]: %s",
+			strings.Join(sockets, ","), err,
+		)
 	}
 	klog.Infoln("cri-o socket:", crioSocket)
 
@@ -50,6 +65,7 @@ func CrioInit() error {
 			DisableCompression: true,
 		},
 	}
+
 	return nil
 }
 

containers/registry.go

@@ -267,16 +267,8 @@ func (r *Registry) handleEvents(ch <-chan ebpftracer.Event) {
 					klog.Infoln("TCP connection error from unknown container", e)
 				}
 			case ebpftracer.EventTypeConnectionClose:
-				if e.Pid != 0 && e.Fd != 0 {
-					if c := r.containersByPid[e.Pid]; c != nil {
-						c.onConnectionClose(e)
-					}
-				} else {
-					for _, c := range r.containersById {
-						if c.onConnectionClose(e) {
-							break
-						}
-					}
+				if c := r.containersByPid[e.Pid]; c != nil {
+					c.onConnectionClose(e)
 				}
 			case ebpftracer.EventTypeTCPRetransmit:
 				srcDst := AddrPair{src: e.SrcAddr, dst: e.DstAddr}

ebpftracer/ebpf/tcp/state.c

@@ -140,19 +140,7 @@ int inet_sock_set_state(void *ctx)
         fd = cid->fd;
     }
     if (args.oldstate == BPF_TCP_ESTABLISHED && (args.newstate == BPF_TCP_FIN_WAIT1 || args.newstate == BPF_TCP_CLOSE_WAIT)) {
-        struct connection_id *cid = bpf_map_lookup_elem(&connection_id_by_socket, &args.skaddr);
-        if (cid) {
-            pid = cid->pid;
-            fd = cid->fd;
-            struct connection *conn = bpf_map_lookup_elem(&active_connections, cid);
-            if (conn) {
-                e.bytes_sent = conn->bytes_sent;
-                e.bytes_received = conn->bytes_received;
-                bpf_map_delete_elem(&active_connections, cid);
-            }
-            bpf_map_delete_elem(&connection_id_by_socket, &args.skaddr);
-        }
-        type = EVENT_TYPE_CONNECTION_CLOSE;
+        bpf_map_delete_elem(&connection_id_by_socket, &args.skaddr);
     }
     if (args.oldstate == BPF_TCP_CLOSE && args.newstate == BPF_TCP_LISTEN) {
         type = EVENT_TYPE_LISTEN_OPEN;
@@ -225,24 +213,20 @@ int sys_enter_close(void *ctx) {
         return 0;
     }
     __u64 id = bpf_get_current_pid_tgid();
-    bpf_map_update_elem(&fd_by_pid_tgid, &id, &args.fd, BPF_ANY);
-    return 0;
-}
-
-SEC("tracepoint/syscalls/sys_exit_close")
-int sys_exit_close(struct trace_event_raw_sys_exit__stub* ctx) {
-    __u64 id = bpf_get_current_pid_tgid();
-    __u64 *fdp = bpf_map_lookup_elem(&fd_by_pid_tgid, &id);
-    if (!fdp) {
-        return 0;
-    }
     struct connection_id cid = {};
     cid.pid = id >> 32;
-    cid.fd = *fdp;
-    bpf_map_delete_elem(&active_connections, &cid);
-    bpf_map_delete_elem(&fd_by_pid_tgid, &id);
+    cid.fd = args.fd;
+    struct connection *conn = bpf_map_lookup_elem(&active_connections, &cid);
+    if (conn) {
+        struct tcp_event e = {};
+        e.type = EVENT_TYPE_CONNECTION_CLOSE;
+        e.pid = cid.pid;
+        e.fd = cid.fd;
+        e.bytes_sent = conn->bytes_sent;
+        e.bytes_received = conn->bytes_received;
+        e.timestamp = conn->timestamp;
+        bpf_perf_event_output(ctx, &tcp_connect_events, BPF_F_CURRENT_CPU, &e, sizeof(e));
+        bpf_map_delete_elem(&active_connections, &cid);
+    }
     return 0;
 }
-
-
-

ebpftracer/tracer.go

@@ -168,12 +168,6 @@ func (t *Tracer) init(ch chan<- Event) error {
 	return nil
 }
 
-func (t *Tracer) GetAndDeleteTCPConnection(pid uint32, fd uint64) (*Connection, error) {
-	id := ConnectionId{FD: fd, PID: pid}
-	conn := &Connection{}
-	return conn, t.collection.Maps["active_connections"].LookupAndDelete(id, conn)
-}
-
 func (t *Tracer) ActiveConnectionsIterator() *ebpf.MapIterator {
 	return t.collection.Maps["active_connections"].Iterate()
 }

go.mod

@@ -10,7 +10,7 @@ require (
 	github.com/containerd/cgroups v1.0.4
 	github.com/containerd/containerd v1.6.26
 	github.com/coreos/go-systemd/v22 v22.5.0
-	github.com/coroot/logparser v1.1.2
+	github.com/coroot/logparser v1.1.5
 	github.com/docker/docker v25.0.6+incompatible
 	github.com/florianl/go-conntrack v0.3.0
 	github.com/go-kit/log v0.2.1

go.sum

@@ -294,8 +294,8 @@ github.com/coreos/pkg v0.0.0-20160727233714-3ac0863d7acf/go.mod h1:E3G3o1h8I7cfc
 github.com/coreos/pkg v0.0.0-20180928190104-399ea9e2e55f/go.mod h1:E3G3o1h8I7cfcXa63jLwjI0eiQQMgzzUDFVpN/nH/eA=
 github.com/coroot/dotnetdiag v1.2.2 h1:PVP/By8o+xhPjfVolJYcjHLbFQInM7pkaD6/otPLc8Q=
 github.com/coroot/dotnetdiag v1.2.2/go.mod h1:veXCMlFzm1yNl7wwJb/ZLxO4WbzhDBoy1VG1XtkH2ls=
-github.com/coroot/logparser v1.1.2 h1:9aH4zIBle14xMHq07YHqVFE2t68k3LE10X2yKHXtJG8=
-github.com/coroot/logparser v1.1.2/go.mod h1:YfYxn9FYBm5GYHHUB4zI22irFAWVDe2bcbOWDHKSmEo=
+github.com/coroot/logparser v1.1.5 h1:gCXeJ0qeRsQWnkK9dOwEiZT3DMjCWp1MTY3ZsPoC3Bk=
+github.com/coroot/logparser v1.1.5/go.mod h1:YfYxn9FYBm5GYHHUB4zI22irFAWVDe2bcbOWDHKSmEo=
 github.com/cpuguy83/go-md2man/v2 v2.0.0-20190314233015-f79a8a8ca69d/go.mod h1:maD7wRr/U5Z6m/iR4s+kqSMx2CaBsrgA7czyZG/E6dU=
 github.com/cpuguy83/go-md2man/v2 v2.0.0/go.mod h1:maD7wRr/U5Z6m/iR4s+kqSMx2CaBsrgA7czyZG/E6dU=
 github.com/cpuguy83/go-md2man/v2 v2.0.2/go.mod h1:tgQtvFlXSQOSOSIRvRPT7W67SCa46tRHOmNcaadrF8o=