Merge pull request #49 from thediveo/develop

Develop

api/types/discovery_test.go

@@ -17,9 +17,10 @@ package types
 import (
 	"encoding/json"
 
+	"github.com/thediveo/lxkns/model"
+
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
-	"github.com/thediveo/lxkns/model"
 	. "github.com/thediveo/lxkns/nstest/gmodel"
 )
 
@@ -38,6 +39,7 @@ var _ = Describe("discovery result JSON", func() {
 			"with-ownership": true,
 			"with-freezer": true,
 			"with-mounts": true,
+			"with-socket-processes": false,
 			"labels": {},
 			"scanned-namespace-types": [
 			  "time",

cmd/lspidns/main_test.go

@@ -93,11 +93,11 @@ read
 			pidnsid.Ino)))
 	})
 
-	It("CLI w/o args renders pid tree", func() {
+	It("CLI -u renders user/pid tree", func() {
 		os.Args = append(os.Args[:1], "-u")
 		out := getstdout.Stdouterr(main)
 		Expect(out).To(MatchRegexp(fmt.Sprintf(`(?m)^user:\[%d\] process .*
-└─ pid:\[%d\] process .*$`,
+[├└]─ pid:\[%d\] process .*$`,
 			initusernsid.Ino, initpidnsid.Ino)))
 		Expect(out).To(MatchRegexp(fmt.Sprintf(`(?m)^   [├└]─ user:\[%d\] process .*
    [│ ]  [├└]─ pid:\[%d\] process .*$`,

decorator/kuhbernetes/cri/decorator_test.go

@@ -17,18 +17,20 @@ package cri
 import (
 	"context"
 	"fmt"
-	"io"
 	"os"
 	"strings"
 	"time"
 
 	"github.com/google/uuid"
-	"github.com/ory/dockertest/v3"
-	"github.com/ory/dockertest/v3/docker"
 	"github.com/thediveo/lxkns/containerizer/whalefriend"
 	"github.com/thediveo/lxkns/decorator/kuhbernetes"
 	"github.com/thediveo/lxkns/model"
 	"github.com/thediveo/lxkns/test/matcher"
+	"github.com/thediveo/morbyd"
+	"github.com/thediveo/morbyd/build"
+	"github.com/thediveo/morbyd/run"
+	"github.com/thediveo/morbyd/session"
+	"github.com/thediveo/morbyd/timestamper"
 	criengine "github.com/thediveo/whalewatcher/engineclient/cri"
 	"github.com/thediveo/whalewatcher/engineclient/cri/test/img"
 	"github.com/thediveo/whalewatcher/test"
@@ -62,7 +64,8 @@ var _ = Describe("k8s (CRI) pods", Ordered, func() {
 		})
 	})
 
-	var providerCntr *dockertest.Resource
+	var sess *morbyd.Session
+	var providerCntr *morbyd.Container
 	var cricl *criengine.Client
 
 	// We build and use the same Docker container for testing our CRI event API
@@ -74,9 +77,13 @@ var _ = Describe("k8s (CRI) pods", Ordered, func() {
 			Skip("needs root")
 		}
 
+		By("creating a new Docker session for testing")
+		sess = Successful(morbyd.NewSession(ctx, session.WithAutoCleaning("lxkns.test=cri")))
+		DeferCleanup(func(ctx context.Context) {
+			sess.Close(ctx)
+		})
+
 		By("spinning up a Docker container with CRI API providers, courtesy of the KinD k8s sig")
-		pool := Successful(dockertest.NewPool("unix:///var/run/docker.sock"))
-		_ = pool.RemoveContainerByName(kindischName)
 		// The necessary container start arguments come from KinD's Docker node
 		// provisioner, see:
 		// https://github.com/kubernetes-sigs/kind/blob/3610f606516ccaa88aa098465d8c13af70937050/pkg/cluster/internal/providers/docker/provision.go#L133
@@ -97,46 +104,35 @@ var _ = Describe("k8s (CRI) pods", Ordered, func() {
 		//   --volume /var
 		//   --volume /lib/modules:/lib/modules:ro
 		//	 kindisch-...
-		Expect(pool.Client.BuildImage(docker.BuildImageOptions{
-			Name:       img.Name,
-			ContextDir: "./test/_kindisch", // sorry, couldn't resist the pun.
-			Dockerfile: "Dockerfile",
-			BuildArgs: []docker.BuildArg{
-				{Name: "KINDEST_BASE_TAG", Value: test.KindestBaseImageTag},
-			},
-			OutputStream: io.Discard,
-		})).To(Succeed())
-		providerCntr = Successful(pool.RunWithOptions(
-			&dockertest.RunOptions{
-				Name:       kindischName,
-				Repository: img.Name,
-				Privileged: true,
-				Mounts: []string{
-					"/var",
-					"/dev/mapper:/dev/mapper",
-					"/lib/modules:/lib/modules:ro",
-				},
-			}, func(hc *docker.HostConfig) {
-				hc.Init = false
-				hc.Tmpfs = map[string]string{
-					"/tmp": "",
-					"/run": "",
-				}
-				hc.Devices = []docker.Device{
-					{PathOnHost: "/dev/fuse"},
-				}
-			}))
-		DeferCleanup(func() {
-			By("removing the CRI API providers Docker container")
-			Expect(pool.Purge(providerCntr)).To(Succeed())
+		Expect(sess.BuildImage(ctx, "./test/_kindisch",
+			build.WithTag(img.Name),
+			build.WithBuildArg("KINDEST_BASE_TAG="+test.KindestBaseImageTag),
+			build.WithOutput(timestamper.New(GinkgoWriter)))).
+			Error().NotTo(HaveOccurred())
+		providerCntr = Successful(sess.Run(ctx, img.Name,
+			run.WithName(kindischName),
+			run.WithAutoRemove(),
+			run.WithPrivileged(),
+			run.WithSecurityOpt("label=disable"),
+			run.WithCgroupnsMode("private"),
+			run.WithVolume("/var"),
+			run.WithVolume("/dev/mapper:/dev/mapper"),
+			run.WithVolume("/lib/modules:/lib/modules:ro"),
+			run.WithTmpfs("/tmp"),
+			run.WithTmpfs("/run"),
+			run.WithDevice("/dev/fuse"),
+			run.WithCombinedOutput(timestamper.New(GinkgoWriter))))
+		DeferCleanup(func(ctx context.Context) {
+			By("removing the test container")
+			providerCntr.Kill(ctx)
 		})
 
 		By("waiting for the CRI API provider to become responsive")
-		Expect(providerCntr.Container.State.Pid).NotTo(BeZero())
+		pid := Successful(providerCntr.PID(ctx))
 		// apipath must not include absolute symbolic links, but already be
 		// properly resolved.
 		endpoint := fmt.Sprintf("/proc/%d/root%s",
-			providerCntr.Container.State.Pid, "/run/containerd/containerd.sock")
+			pid, "/run/containerd/containerd.sock")
 		Eventually(func() error {
 			var err error
 			cricl, err = criengine.New(endpoint, criengine.WithTimeout(1*time.Second))
@@ -154,7 +150,7 @@ var _ = Describe("k8s (CRI) pods", Ordered, func() {
 		defer cancel()
 		criw := watcher.New(criengine.NewCRIWatcher(
 			Successful(criengine.New(strings.TrimPrefix(cricl.Address(), "unix://"))),
-			criengine.WithPID(providerCntr.Container.State.Pid)), nil)
+			criengine.WithPID(Successful(providerCntr.PID(ctx)))), nil)
 		cizer := whalefriend.New(ctx, []watcher.Watcher{criw})
 		defer cizer.Close()
 		Eventually(criw.Ready()).Within(5 * time.Second).Should(BeClosed())

deployments/lxkns/Dockerfile

@@ -14,9 +14,9 @@
 # done executing "make deploy" with the current working directory set to this
 # repository's root directory.
 
-ARG ALPINE_VERSION=3.19
-ARG ALPINE_PATCH=0
-ARG GO_VERSION=1.21.6
+ARG ALPINE_VERSION=3.20
+ARG ALPINE_PATCH=1
+ARG GO_VERSION=1.22.4
 ARG NODE_VERSION=21
 
 # 0th stage: https://github.com/tonistiigi/xx/blob/master/README.md

discover/discover.go

@@ -42,8 +42,19 @@ type Result struct {
 	PIDMap            model.PIDMapper        `json:"-"` // optional PID translator.
 	Mounts            NamespacedMountPathMap // per mount-namespace mount paths and mount points.
 	Containers        model.Containers       // all alive containers found
+	SocketProcessMap  SocketProcesses        // optional socket inode number to process(es) mapping
 }
 
+// SocketProcesses maps socket inode numbers to processes that have open file
+// descriptors for specific sockets.
+//
+// As it turned out over time, there are multiple lxkns API users that otherwise
+// repeatedly scan the open file descriptors of processes for sockets in order
+// to gather their inode numbers, so in the sense of DRY we offer this
+// information with a single scan we need to do anyway in case discovering
+// network namespaces from sockets.
+type SocketProcesses map[uint64][]model.PIDType
+
 // SortNamespaces returns a sorted copy of a list of namespaces. The
 // namespaces are sorted by their namespace ids in ascending order.
 func SortNamespaces(nslist []model.Namespace) []model.Namespace {

discover/discovery_fd.go

@@ -35,22 +35,40 @@ import (
 	"golang.org/x/sys/unix"
 )
 
-// discoverFromFd discovers namespaces from process file descriptors referencing
-// them either directly or via socket fds. Since file descriptors are per
-// process only, but not per task/thread, it sufficies to only iterate the
-// process fd entries, leaving out the copies in the task fd entries.
+// discoverFromFd discovers (1) namespaces from open file descriptors
+// referencing them either directly or sockets that are attached to them, as
+// well as (2) the socket-to-processes mapping in a single run. This way we
+// avoid DRY of repeated open fd socket scanning.
+//
+// Please note that scanning file descriptors for namespaces and sockets
+// automatically opts into discovering the socket-to-processes mapping, as this
+// is a byproduct anyway.
+//
+// Since file descriptors are per process only, but not per task/thread, it
+// sufficies to only iterate the process fd entries, leaving out the copies in
+// the task fd entries.
 func discoverFromFd(t species.NamespaceType, procfs string, result *Result) {
-	if !result.Options.ScanFds {
-		log.Infof("skipping discovery of fd-referenced namespaces")
+	if !result.Options.ScanFds && !result.Options.DiscoverSocketProcesses {
+		log.Infof("skipping discovery of fd-referenced namespaces and socket processes")
 		return
 	}
-	log.Debugf("discovering fd-referenced namespaces...")
+	switch {
+	case result.Options.ScanFds:
+		log.Debugf("discovering fd-referenced namespaces and socket processes...")
+	default:
+		log.Debugf("discovering socket processes...")
+	}
 	scanFd(t, procfs, false, result)
 }
 
+const socketPrefix = "socket:["
+const socketPrefixLen = len(socketPrefix)
+
 // namespaceFromFd is discoverFromFd with special test harness handling enabled
 // or disabled.
 func scanFd(_ species.NamespaceType, procfs string, fakeprocfs bool, result *Result) {
+	result.SocketProcessMap = SocketProcesses{}
+	scanFds := result.Options.ScanFds
 	// Iterate over all known processes, and then over all of their open file
 	// descriptors. The /proc filesystem will give us the required
 	// information.
@@ -98,16 +116,36 @@ func scanFd(_ species.NamespaceType, procfs string, fakeprocfs bool, result *Res
 			}
 			var nsid species.NamespaceID
 			var nstype species.NamespaceType
-			if strings.HasPrefix(fdtarget, "socket:[") {
-				// It's a socket ... and we want to query it using an ioctl for
-				// the network namespace it is connected to.
+			if strings.HasPrefix(fdtarget, socketPrefix) {
+				// It's a socket so we note down the relationship between the
+				// socket's inode number and this process in any case, as this
+				// is a byproduct of trying to find the socket's network
+				// namespace.
+				l := len(fdtarget)
+				if l <= socketPrefixLen {
+					continue
+				}
+				ino, err := strconv.ParseUint(fdtarget[8:l-1], 10, 64)
+				if err != nil {
+					continue
+				}
+				result.SocketProcessMap[ino] = append(result.SocketProcessMap[ino], pid)
+				if !scanFds {
+					continue
+				}
+				// So the calling explorer really wants to discover network
+				// namespaces from sockets. If we haven't done yet for this
+				// process, get a PID fd so we can later duplicate the
+				// processes's fd into our process for further inspection.
 				if pidfd <= 0 {
 					pidfd, err = unix.PidfdOpen(int(pid), 0)
 					if err != nil {
 						continue
 					}
 				}
 				nsid, nstype = namespaceOfSocket(pidfd, fdentry.Name())
+			} else if !scanFds {
+				continue
 			} else {
 				nsid, nstype = namespaceFromLink(procfdpath, fdtarget, fakeprocfs)
 			}
@@ -138,7 +176,10 @@ func scanFd(_ species.NamespaceType, procfs string, fakeprocfs bool, result *Res
 			pidfd = 0
 		}
 	}
-	log.Infof("discovered %s", plural.Elements(total, "fd-referenced namespaces"))
+	if result.Options.ScanFds {
+		log.Infof("discovered %s", plural.Elements(total, "fd-referenced namespaces"))
+	}
+	log.Infof("discovered %s", plural.Elements(len(result.SocketProcessMap), "sockets"))
 }
 
 // namespaceOfSocket returns the network namespace a particular socket fd (of

discover/discovery_fd_test.go

@@ -16,6 +16,7 @@ package discover
 
 import (
 	"os"
+	"sync"
 	"time"
 
 	"github.com/thediveo/lxkns/model"
@@ -29,7 +30,7 @@ import (
 	. "github.com/onsi/gomega"
 	. "github.com/onsi/gomega/gleak"
 	. "github.com/thediveo/fdooze"
-	. "github.com/thediveo/once"
+	. "github.com/thediveo/success"
 )
 
 var _ = Describe("Discover from fds", func() {
@@ -105,20 +106,41 @@ read # wait for test to proceed()
 
 		By("creating a transient new network namespace we only keep a socket connected to")
 		netnsFd := netns.NewTransient()
-		closeOnceNetnsFd := Once(func() { unix.Close(netnsFd) })
-		defer closeOnceNetnsFd.Do()
+		closeNetnsFd := sync.OnceFunc(func() { unix.Close(netnsFd) })
+		defer closeNetnsFd()
 
 		netnsino := netns.Ino(netnsFd)
 
 		nlh := netns.NewNetlinkHandle(netnsFd)
 		defer nlh.Close()
 
 		By("keeping only a socket as the last reference to the transient network namespace")
-		closeOnceNetnsFd.Do()
+		closeNetnsFd()
 
 		By("discovering the transient network namespace from the RTNETLINK socket")
 		allns := Namespaces(FromFds())
-		Expect(allns.Namespaces[model.NetNS]).To(HaveKey(species.NamespaceIDfromInode(netnsino)))
+		Expect(allns.Namespaces[model.NetNS]).To(
+			HaveKey(species.NamespaceIDfromInode(netnsino)))
+	})
+
+	It("discovers the socket-to-process mapping", func() {
+		if os.Getuid() != 0 {
+			Skip("needs root")
+		}
+
+		sockfd := Successful(unix.Socket(unix.AF_INET, unix.SOCK_DGRAM, 0))
+		defer unix.Close(sockfd)
+		var sockstat unix.Stat_t
+		Expect(unix.Fstat(sockfd, &sockstat)).To(Succeed())
+
+		By("requesting scanning fds for socket network namespaces")
+		allns := Namespaces(FromFds())
+		Expect(allns.SocketProcessMap).To(HaveKeyWithValue(
+			sockstat.Ino, ConsistOf(model.PIDType(os.Getpid()))))
+
+		allns = Namespaces(WithSocketProcesses())
+		Expect(allns.SocketProcessMap).To(HaveKeyWithValue(
+			sockstat.Ino, ConsistOf(model.PIDType(os.Getpid()))))
 	})
 
 })