ebpf: handle fd_install

Author: alban

probe/endpoint/connection_tracker.go

@@ -171,7 +171,8 @@ func (t *connectionTracker) performWalkProc(rpt *report.Report, hostNodeID strin
 // once to initialize ebpfTracker
 func (t *connectionTracker) getInitialState() {
 	var processCache *process.CachingWalker
-	processCache = process.NewCachingWalker(process.NewWalker(t.conf.ProcRoot))
+	walker := process.NewWalker(t.conf.ProcRoot, true)
+	processCache = process.NewCachingWalker(walker)
 	processCache.Tick()
 
 	scanner := procspy.NewSyncConnectionScanner(processCache)
@@ -194,7 +195,14 @@ func (t *connectionTracker) getInitialState() {
 	}
 	scanner.Stop()
 
-	t.ebpfTracker.feedInitialConnections(conns, seenTuples, report.MakeHostNodeID(t.conf.HostID))
+	processesWaitingInAccept := []int{}
+	processCache.Walk(func(p, prev process.Process) {
+		if p.IsWaitingInAccept {
+			processesWaitingInAccept = append(processesWaitingInAccept, p.PID)
+		}
+	})
+
+	t.ebpfTracker.feedInitialConnections(conns, seenTuples, processesWaitingInAccept, report.MakeHostNodeID(t.conf.HostID))
 }
 
 func (t *connectionTracker) performEbpfTrack(rpt *report.Report, hostNodeID string) error {

probe/endpoint/ebpf.go

@@ -1,14 +1,18 @@
 package endpoint
 
 import (
+	"bytes"
 	"fmt"
 	"regexp"
 	"strconv"
 	"sync"
+	"syscall"
 
 	log "github.com/Sirupsen/logrus"
+	"github.com/weaveworks/common/fs"
 	"github.com/weaveworks/scope/probe/endpoint/procspy"
 	"github.com/weaveworks/scope/probe/host"
+	"github.com/weaveworks/scope/probe/process"
 	"github.com/weaveworks/tcptracer-bpf/pkg/tracer"
 )
 
@@ -23,7 +27,7 @@ type ebpfConnection struct {
 type eventTracker interface {
 	handleConnection(ev tracer.EventType, tuple fourTuple, pid int, networkNamespace string)
 	walkConnections(f func(ebpfConnection))
-	feedInitialConnections(ci procspy.ConnIter, seenTuples map[string]fourTuple, hostNodeID string)
+	feedInitialConnections(ci procspy.ConnIter, seenTuples map[string]fourTuple, processesWaitingInAccept []int, hostNodeID string)
 	isReadyToHandleConnections() bool
 	isDead() bool
 	stop()
@@ -111,8 +115,12 @@ func tcpEventCbV4(e tracer.TcpV4) {
 
 	lastTimestampV4 = e.Timestamp
 
-	tuple := fourTuple{e.SAddr.String(), e.DAddr.String(), e.SPort, e.DPort}
-	ebpfTracker.handleConnection(e.Type, tuple, int(e.Pid), strconv.Itoa(int(e.NetNS)))
+	if e.Type == tracer.EventFdInstall {
+		ebpfTracker.handleFdInstall(e.Type, int(e.Pid), int(e.Fd))
+	} else {
+		tuple := fourTuple{e.SAddr.String(), e.DAddr.String(), e.SPort, e.DPort}
+		ebpfTracker.handleConnection(e.Type, tuple, int(e.Pid), strconv.Itoa(int(e.NetNS)))
+	}
 }
 
 func tcpEventCbV6(e tracer.TcpV6) {
@@ -125,6 +133,84 @@ func lostCb(count uint64) {
 	ebpfTracker.stop()
 }
 
+func tupleFromPidFd(pid int, fd int) (tuple fourTuple, netns string, ok bool) {
+	// read /proc/$pid/ns/net
+	//
+	// probe/endpoint/procspy/proc_linux.go supports Linux < 3.8 but we
+	// don't need that here since ebpf-enabled kernels will be > 3.8
+	netNamespacePath := fmt.Sprintf("/proc/%d/ns/net", pid)
+	var statNsFile syscall.Stat_t
+	err := fs.Stat(netNamespacePath, &statNsFile)
+	if err != nil {
+		log.Debugf("proc file %q disappeared before we could read it", netNamespacePath)
+		return fourTuple{}, "", false
+	}
+	netns = fmt.Sprintf("%d", statNsFile.Ino)
+
+	// find /proc/$pid/fd/$fd's ino
+	fdFilename := fmt.Sprintf("/proc/%d/fd/%d", pid, fd)
+	var statFdFile syscall.Stat_t
+	err = fs.Stat(fdFilename, &statFdFile)
+	if err != nil {
+		log.Debugf("proc file %q disappeared before we could read it", fdFilename)
+		return fourTuple{}, "", false
+	}
+
+	if statFdFile.Mode&syscall.S_IFMT != syscall.S_IFSOCK {
+		log.Errorf("file %q is not a socket", fdFilename)
+		return fourTuple{}, "", false
+	}
+	ino := statFdFile.Ino
+
+	// read /proc/pid/net/tcp
+	buf := bytes.NewBuffer(make([]byte, 0, 5000))
+	tcpFilename := fmt.Sprintf("/proc/%d/net/tcp", pid)
+	f, err := fs.Open(tcpFilename)
+	if err != nil {
+		log.Debugf("proc file %q disappeared before we could read it", tcpFilename)
+		return fourTuple{}, "", false
+	}
+	defer f.Close()
+	_, err = buf.ReadFrom(f)
+	if err != nil {
+		log.Debugf("proc file %q disappeared before we could read it", tcpFilename)
+		return fourTuple{}, "", false
+	}
+
+	// find /proc/$pid/fd/$fd's ino in /proc/pid/net/tcp
+	pn := procspy.NewProcNet(buf.Bytes())
+	for {
+		n := pn.Next()
+		if n == nil {
+			log.Debugf("connection for proc file %q disappeared before we could get it", fdFilename)
+			break
+		}
+		if n.Inode == ino {
+			return fourTuple{n.LocalAddress.String(), n.RemoteAddress.String(), n.LocalPort, n.RemotePort}, netns, true
+		}
+	}
+
+	return fourTuple{}, "", false
+}
+
+func (t *EbpfTracker) handleFdInstall(ev tracer.EventType, pid int, fd int) {
+	tuple, netns, ok := tupleFromPidFd(pid, fd)
+	log.Debugf("EbpfTracker: got fd-install event: pid=%d fd=%d -> tuple=%s netns=%s ok=%v", pid, fd, tuple, netns, ok)
+	if !ok {
+		return
+	}
+	conn := ebpfConnection{
+		incoming:         true,
+		tuple:            tuple,
+		pid:              pid,
+		networkNamespace: netns,
+	}
+	t.openConnections[tuple.String()] = conn
+	if !process.IsProcInAccept("/proc", strconv.Itoa(pid)) {
+		t.tracer.RemoveFdInstallWatcher(uint32(pid))
+	}
+}
+
 func (t *EbpfTracker) handleConnection(ev tracer.EventType, tuple fourTuple, pid int, networkNamespace string) {
 	t.Lock()
 	defer t.Unlock()
@@ -160,6 +246,8 @@ func (t *EbpfTracker) handleConnection(ev tracer.EventType, tuple fourTuple, pid
 		} else {
 			log.Debugf("EbpfTracker: unmatched close event: %s pid=%d netns=%s", tuple.String(), pid, networkNamespace)
 		}
+	default:
+		log.Debugf("EbpfTracker: unknown event: %s (%d)", ev, ev)
 	}
 }
 
@@ -178,7 +266,7 @@ func (t *EbpfTracker) walkConnections(f func(ebpfConnection)) {
 	t.closedConnections = t.closedConnections[:0]
 }
 
-func (t *EbpfTracker) feedInitialConnections(conns procspy.ConnIter, seenTuples map[string]fourTuple, hostNodeID string) {
+func (t *EbpfTracker) feedInitialConnections(conns procspy.ConnIter, seenTuples map[string]fourTuple, processesWaitingInAccept []int, hostNodeID string) {
 	t.readyToHandleConnections = true
 	for conn := conns.Next(); conn != nil; conn = conns.Next() {
 		var (
@@ -204,6 +292,10 @@ func (t *EbpfTracker) feedInitialConnections(conns procspy.ConnIter, seenTuples
 			t.handleConnection(tracer.EventAccept, tuple, int(conn.Proc.PID), namespaceID)
 		}
 	}
+	for _, p := range processesWaitingInAccept {
+		t.tracer.AddFdInstallWatcher(uint32(p))
+		log.Debugf("EbpfTracker: install fd-install watcher: pid=%d", p)
+	}
 }
 
 func (t *EbpfTracker) isReadyToHandleConnections() bool {

probe/endpoint/procspy/proc_internal_test.go

@@ -62,7 +62,7 @@ func TestWalkProcPid(t *testing.T) {
 	defer fs_hook.Restore()
 
 	buf := bytes.Buffer{}
-	walker := process.NewWalker(procRoot)
+	walker := process.NewWalker(procRoot, false)
 	ticker := time.NewTicker(time.Millisecond)
 	defer ticker.Stop()
 	pWalker := newPidWalker(walker, ticker.C, 1)

probe/endpoint/procspy/procnet.go

@@ -63,12 +63,12 @@ again:
 
 	p.c.LocalAddress, p.c.LocalPort = scanAddressNA(local, &p.bytesLocal)
 	p.c.RemoteAddress, p.c.RemotePort = scanAddressNA(remote, &p.bytesRemote)
-	p.c.inode = parseDec(inode)
+	p.c.Inode = parseDec(inode)
 	p.b = nextLine(b)
-	if _, alreadySeen := p.seen[p.c.inode]; alreadySeen {
+	if _, alreadySeen := p.seen[p.c.Inode]; alreadySeen {
 		goto again
 	}
-	p.seen[p.c.inode] = struct{}{}
+	p.seen[p.c.Inode] = struct{}{}
 	return &p.c
 }
 

probe/endpoint/procspy/procnet_internal_test.go

@@ -20,28 +20,28 @@ func TestProcNet(t *testing.T) {
 			LocalPort:     0xa6c0,
 			RemoteAddress: net.IP([]byte{0, 0, 0, 0}),
 			RemotePort:    0x0,
-			inode:         5107,
+			Inode:         5107,
 		},
 		{
 			LocalAddress:  net.IP([]byte{0, 0, 0, 0}),
 			LocalPort:     0x006f,
 			RemoteAddress: net.IP([]byte{0, 0, 0, 0}),
 			RemotePort:    0x0,
-			inode:         5084,
+			Inode:         5084,
 		},
 		{
 			LocalAddress:  net.IP([]byte{0x7f, 0x0, 0x0, 0x01}),
 			LocalPort:     0x0019,
 			RemoteAddress: net.IP([]byte{0, 0, 0, 0}),
 			RemotePort:    0x0,
-			inode:         10550,
+			Inode:         10550,
 		},
 		{
 			LocalAddress:  net.IP([]byte{0x2e, 0xf6, 0x2c, 0xa1}),
 			LocalPort:     0xe4d7,
 			RemoteAddress: net.IP([]byte{0xc0, 0x1e, 0xfc, 0x57}),
 			RemotePort:    0x01bb,
-			inode:         639474,
+			Inode:         639474,
 		},
 	}
 	for i := 0; i < 4; i++ {
@@ -73,7 +73,7 @@ func TestTransport6(t *testing.T) {
 			RemoteAddress: net.IP(make([]byte, 16)),
 			RemotePort:    0x0,
 			// uid:           0,
-			inode: 23661201,
+			Inode: 23661201,
 		},
 		{
 			// state: 1,
@@ -92,7 +92,7 @@ func TestTransport6(t *testing.T) {
 			}),
 			RemotePort: 0x01bb,
 			// uid:        1000,
-			inode: 36856710,
+			Inode: 36856710,
 		},
 	}
 
@@ -148,7 +148,7 @@ func TestProcNetFiltersDuplicates(t *testing.T) {
 		LocalPort:     0xa6c0,
 		RemoteAddress: net.IP([]byte{0, 0, 0, 0}),
 		RemotePort:    0x0,
-		inode:         5107,
+		Inode:         5107,
 	}
 	have := p.Next()
 	want := expected

probe/endpoint/procspy/spy.go

@@ -22,7 +22,7 @@ type Connection struct {
 	LocalPort     uint16
 	RemoteAddress net.IP
 	RemotePort    uint16
-	inode         uint64
+	Inode         uint64
 	Proc          Proc
 }
 

probe/endpoint/procspy/spy_linux.go

@@ -26,7 +26,7 @@ func (c *pnConnIter) Next() *Connection {
 		bufPool.Put(c.buf)
 		return nil
 	}
-	if proc, ok := c.procs[n.inode]; ok {
+	if proc, ok := c.procs[n.Inode]; ok {
 		n.Proc = *proc
 	}
 	return n

probe/endpoint/procspy/spy_linux_internal_test.go

@@ -14,7 +14,7 @@ import (
 func TestLinuxConnections(t *testing.T) {
 	fs_hook.Mock(mockFS)
 	defer fs_hook.Restore()
-	scanner := NewConnectionScanner(process.NewWalker("/proc"))
+	scanner := NewConnectionScanner(process.NewWalker("/proc", false))
 	defer scanner.Stop()
 
 	// let the background scanner finish its first pass
@@ -30,7 +30,7 @@ func TestLinuxConnections(t *testing.T) {
 		LocalPort:     42688,
 		RemoteAddress: net.ParseIP("0.0.0.0").To4(),
 		RemotePort:    0,
-		inode:         5107,
+		Inode:         5107,
 		Proc: Proc{
 			PID:  1,
 			Name: "foo",

probe/process/walker.go

@@ -4,15 +4,16 @@ import "sync"
 
 // Process represents a single process.
 type Process struct {
-	PID, PPID      int
-	Name           string
-	Cmdline        string
-	Threads        int
-	Jiffies        uint64
-	RSSBytes       uint64
-	RSSBytesLimit  uint64
-	OpenFilesCount int
-	OpenFilesLimit uint64
+	PID, PPID         int
+	Name              string
+	Cmdline           string
+	Threads           int
+	Jiffies           uint64
+	RSSBytes          uint64
+	RSSBytesLimit     uint64
+	OpenFilesCount    int
+	OpenFilesLimit    uint64
+	IsWaitingInAccept bool
 }
 
 // Walker is something that walks the /proc directory