From f792d893c643cfd79c60f254c300cd73293a3497 Mon Sep 17 00:00:00 2001 From: iemAnshuman Date: Tue, 4 Aug 2026 02:20:29 +0530 Subject: [PATCH 01/20] add the host pid broker protocol Signed-off-by: iemAnshuman --- .../nvinternal/hostpid/protocol.go | 39 ++++++++++++ .../nvinternal/hostpid/protocol_test.go | 62 +++++++++++++++++++ 2 files changed, 101 insertions(+) create mode 100644 pkg/device-plugin/nvidiadevice/nvinternal/hostpid/protocol.go create mode 100644 pkg/device-plugin/nvidiadevice/nvinternal/hostpid/protocol_test.go diff --git a/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/protocol.go b/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/protocol.go new file mode 100644 index 0000000000..33e5bc3313 --- /dev/null +++ b/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/protocol.go @@ -0,0 +1,39 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * Copyright (c) 2026 The HAMi Authors. + */ + +package hostpid + +import "encoding/binary" + +const ( + protocolVersion uint16 = 1 + commandGetPID uint16 = 1 + + statusOK uint16 = 0 + statusInvalidRequest uint16 = 1 + + requestSize = 8 + responseSize = 12 +) + +var protocolMagic = [4]byte{'H', 'P', 'I', 'D'} + +func validRequest(request []byte) bool { + return len(request) == requestSize && + string(request[:4]) == string(protocolMagic[:]) && + binary.BigEndian.Uint16(request[4:6]) == protocolVersion && + binary.BigEndian.Uint16(request[6:8]) == commandGetPID +} + +func makeResponse(status uint16, pid uint32) [responseSize]byte { + var response [responseSize]byte + + copy(response[:4], protocolMagic[:]) + binary.BigEndian.PutUint16(response[4:6], protocolVersion) + binary.BigEndian.PutUint16(response[6:8], status) + binary.BigEndian.PutUint32(response[8:12], pid) + return response +} diff --git a/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/protocol_test.go b/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/protocol_test.go new file mode 100644 index 0000000000..96c8866e5c --- /dev/null +++ b/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/protocol_test.go @@ -0,0 +1,62 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * Copyright (c) 2026 The HAMi Authors. + */ + +package hostpid + +import ( + "encoding/binary" + "testing" +) + +func TestValidRequest(t *testing.T) { + valid := []byte{'H', 'P', 'I', 'D', 0, 1, 0, 1} + if !validRequest(valid) { + t.Fatal("valid request was rejected") + } + + tests := map[string][]byte{ + "short": valid[:7], + "long": append(append([]byte{}, valid...), 0), + "magic": {'B', 'A', 'D', '!', 0, 1, 0, 1}, + "version": {'H', 'P', 'I', 'D', 0, 2, 0, 1}, + "command": {'H', 'P', 'I', 'D', 0, 1, 0, 2}, + "zero command": {'H', 'P', 'I', 'D', 0, 1, 0, 0}, + } + for name, request := range tests { + t.Run(name, func(t *testing.T) { + if validRequest(request) { + t.Fatal("invalid request was accepted") + } + }) + } +} + +func TestMakeResponse(t *testing.T) { + response := makeResponse(statusOK, 0x01020304) + + if string(response[:4]) != "HPID" { + t.Fatalf("unexpected magic %q", response[:4]) + } + if got := binary.BigEndian.Uint16(response[4:6]); got != protocolVersion { + t.Fatalf("unexpected version %d", got) + } + if got := binary.BigEndian.Uint16(response[6:8]); got != statusOK { + t.Fatalf("unexpected status %d", got) + } + if got := binary.BigEndian.Uint32(response[8:12]); got != 0x01020304 { + t.Fatalf("unexpected PID %#x", got) + } +} + +func TestMakeErrorResponse(t *testing.T) { + response := makeResponse(statusInvalidRequest, 0) + if got := binary.BigEndian.Uint16(response[6:8]); got != statusInvalidRequest { + t.Fatalf("unexpected status %d", got) + } + if got := binary.BigEndian.Uint32(response[8:12]); got != 0 { + t.Fatalf("unexpected PID %d", got) + } +} From f4e525800cbf79d6c07683833621d370c0e5c4a2 Mon Sep 17 00:00:00 2001 From: iemAnshuman Date: Tue, 4 Aug 2026 02:30:54 +0530 Subject: [PATCH 02/20] serve host pids over unix sockets Signed-off-by: iemAnshuman --- go.mod | 2 +- .../nvinternal/hostpid/broker_linux.go | 344 +++++++++++++++ .../nvinternal/hostpid/broker_linux_test.go | 396 ++++++++++++++++++ .../nvinternal/hostpid/broker_unsupported.go | 27 ++ .../nvidiadevice/nvinternal/hostpid/config.go | 21 + 5 files changed, 789 insertions(+), 1 deletion(-) create mode 100644 pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_linux.go create mode 100644 pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_linux_test.go create mode 100644 pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_unsupported.go create mode 100644 pkg/device-plugin/nvidiadevice/nvinternal/hostpid/config.go diff --git a/go.mod b/go.mod index 6ee832f562..ba687a2039 100644 --- a/go.mod +++ b/go.mod @@ -22,6 +22,7 @@ require ( github.com/stretchr/testify v1.11.1 github.com/urfave/cli/v2 v2.27.7 golang.org/x/net v0.57.0 + golang.org/x/sys v0.47.0 golang.org/x/term v0.45.0 golang.org/x/tools v0.48.0 google.golang.org/grpc v1.83.0 @@ -81,7 +82,6 @@ require ( golang.org/x/mod v0.38.0 // indirect golang.org/x/oauth2 v0.36.0 // indirect golang.org/x/sync v0.22.0 // indirect - golang.org/x/sys v0.47.0 // indirect golang.org/x/telemetry v0.0.0-20260708182218-49f421fb7959 // indirect golang.org/x/text v0.40.0 // indirect golang.org/x/time v0.14.0 // indirect diff --git a/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_linux.go b/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_linux.go new file mode 100644 index 0000000000..b284298b00 --- /dev/null +++ b/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_linux.go @@ -0,0 +1,344 @@ +//go:build linux + +/* + * SPDX-License-Identifier: Apache-2.0 + * + * Copyright (c) 2026 The HAMi Authors. + */ + +package hostpid + +import ( + "errors" + "fmt" + "io" + "net" + "os" + "path/filepath" + "sync" + "sync/atomic" + "syscall" + "time" + + "golang.org/x/sys/unix" +) + +const ( + transactionTimeout = 500 * time.Millisecond + activeProbeTimeout = 50 * time.Millisecond + maxHandlers = 512 + serverDirectoryMode = 0o711 + serverSocketMode = 0o666 + serverLockMode = 0o600 +) + +type socketIdentity struct { + device uint64 + inode uint64 +} + +type Broker struct { + listener *net.UnixListener + socketPath string + socket socketIdentity + lockFile *os.File + handlerSlots chan struct{} + handlerWG sync.WaitGroup + handlerMu sync.Mutex + closing atomic.Bool + closeOnce sync.Once + closeErr error +} + +func ListenDefault() (*Broker, error) { + if os.Geteuid() != 0 { + return nil, errors.New("the host PID broker must run as root") + } + return listen(ServerSocketPath, 0) +} + +func listen(socketPath string, ownerUID int) (*Broker, error) { + directory := filepath.Dir(socketPath) + if err := prepareDirectory(directory, ownerUID); err != nil { + return nil, err + } + + lockFile, err := acquireLock(directory, ownerUID) + if err != nil { + return nil, err + } + releaseLock := true + defer func() { + if releaseLock { + _ = unix.Flock(int(lockFile.Fd()), unix.LOCK_UN) + _ = lockFile.Close() + } + }() + + if err := removeStaleSocket(socketPath, ownerUID); err != nil { + return nil, err + } + address := &net.UnixAddr{Name: socketPath, Net: "unix"} + listener, err := net.ListenUnix("unix", address) + if err != nil { + return nil, fmt.Errorf("listen on host PID broker socket: %w", err) + } + listener.SetUnlinkOnClose(false) + cleanupListener := true + defer func() { + if cleanupListener { + _ = listener.Close() + _ = os.Remove(socketPath) + } + }() + + if err := os.Chmod(socketPath, serverSocketMode); err != nil { + return nil, fmt.Errorf("set host PID broker socket mode: %w", err) + } + identity, err := readSocketIdentity(socketPath, ownerUID) + if err != nil { + return nil, err + } + + cleanupListener = false + releaseLock = false + return &Broker{ + listener: listener, + socketPath: socketPath, + socket: identity, + lockFile: lockFile, + handlerSlots: make(chan struct{}, maxHandlers), + }, nil +} + +func prepareDirectory(directory string, ownerUID int) error { + if err := os.MkdirAll(directory, serverDirectoryMode); err != nil { + return fmt.Errorf("create host PID broker directory: %w", err) + } + info, err := os.Lstat(directory) + if err != nil { + return fmt.Errorf("inspect host PID broker directory: %w", err) + } + stat, ok := info.Sys().(*syscall.Stat_t) + if !ok || !info.IsDir() || info.Mode()&os.ModeSymlink != 0 { + return errors.New("host PID broker directory is not a real directory") + } + if int(stat.Uid) != ownerUID { + return fmt.Errorf("host PID broker directory owner is %d, want %d", + stat.Uid, ownerUID) + } + if err := os.Chmod(directory, serverDirectoryMode); err != nil { + return fmt.Errorf("set host PID broker directory mode: %w", err) + } + return nil +} + +func acquireLock(directory string, ownerUID int) (*os.File, error) { + lockPath := filepath.Join(directory, "broker.lock") + fd, err := unix.Open(lockPath, + unix.O_CREAT|unix.O_RDWR|unix.O_CLOEXEC|unix.O_NOFOLLOW, + serverLockMode) + if err != nil { + return nil, fmt.Errorf("open host PID broker lock: %w", err) + } + lockFile := os.NewFile(uintptr(fd), lockPath) + if lockFile == nil { + _ = unix.Close(fd) + return nil, errors.New("open host PID broker lock file") + } + if err := unix.Flock(fd, unix.LOCK_EX|unix.LOCK_NB); err != nil { + _ = lockFile.Close() + return nil, fmt.Errorf("lock host PID broker directory: %w", err) + } + + info, err := lockFile.Stat() + if err != nil { + _ = unix.Flock(fd, unix.LOCK_UN) + _ = lockFile.Close() + return nil, fmt.Errorf("inspect host PID broker lock: %w", err) + } + stat, ok := info.Sys().(*syscall.Stat_t) + if !ok || !info.Mode().IsRegular() || int(stat.Uid) != ownerUID || + info.Mode().Perm()&0o077 != 0 { + _ = unix.Flock(fd, unix.LOCK_UN) + _ = lockFile.Close() + return nil, errors.New("host PID broker lock is not trusted") + } + return lockFile, nil +} + +func removeStaleSocket(socketPath string, ownerUID int) error { + info, err := os.Lstat(socketPath) + if errors.Is(err, os.ErrNotExist) { + return nil + } + if err != nil { + return fmt.Errorf("inspect host PID broker socket: %w", err) + } + stat, ok := info.Sys().(*syscall.Stat_t) + if !ok || info.Mode()&os.ModeSocket == 0 { + return errors.New("host PID broker socket path is not a socket") + } + if int(stat.Uid) != ownerUID { + return fmt.Errorf("host PID broker socket owner is %d, want %d", + stat.Uid, ownerUID) + } + + connection, dialErr := net.DialTimeout("unix", socketPath, + activeProbeTimeout) + if dialErr == nil { + _ = connection.Close() + return errors.New("another host PID broker is already listening") + } + if !errors.Is(dialErr, syscall.ECONNREFUSED) && + !errors.Is(dialErr, os.ErrNotExist) { + return fmt.Errorf("probe existing host PID broker socket: %w", dialErr) + } + if err := os.Remove(socketPath); err != nil && + !errors.Is(err, os.ErrNotExist) { + return fmt.Errorf("remove stale host PID broker socket: %w", err) + } + return nil +} + +func readSocketIdentity(socketPath string, ownerUID int) (socketIdentity, error) { + info, err := os.Lstat(socketPath) + if err != nil { + return socketIdentity{}, fmt.Errorf( + "inspect new host PID broker socket: %w", err) + } + stat, ok := info.Sys().(*syscall.Stat_t) + if !ok || info.Mode()&os.ModeSocket == 0 || int(stat.Uid) != ownerUID { + return socketIdentity{}, errors.New("new host PID broker socket is not trusted") + } + return socketIdentity{device: uint64(stat.Dev), inode: stat.Ino}, nil +} + +func (broker *Broker) Serve() error { + for { + connection, err := broker.listener.AcceptUnix() + if err != nil { + if broker.closing.Load() { + return nil + } + return fmt.Errorf("accept host PID broker connection: %w", err) + } + if !broker.startHandler(connection) { + _ = connection.Close() + } + } +} + +func (broker *Broker) startHandler(connection *net.UnixConn) bool { + broker.handlerMu.Lock() + defer broker.handlerMu.Unlock() + if broker.closing.Load() { + return false + } + select { + case broker.handlerSlots <- struct{}{}: + broker.handlerWG.Add(1) + go broker.handle(connection) + return true + default: + return false + } +} + +func (broker *Broker) handle(connection *net.UnixConn) { + defer func() { + _ = connection.Close() + <-broker.handlerSlots + broker.handlerWG.Done() + }() + _ = connection.SetDeadline(time.Now().Add(transactionTimeout)) + + request := make([]byte, requestSize) + if _, err := io.ReadFull(connection, request); err != nil { + return + } + if !validRequest(request) { + response := makeResponse(statusInvalidRequest, 0) + writeResponse(connection, response) + return + } + + pid, err := peerPID(connection) + if err != nil || pid <= 0 { + return + } + response := makeResponse(statusOK, uint32(pid)) + writeResponse(connection, response) +} + +func writeResponse(connection *net.UnixConn, + response [responseSize]byte) { + written := 0 + for written < len(response) { + count, err := connection.Write(response[written:]) + if err != nil || count == 0 { + return + } + written += count + } +} + +func peerPID(connection *net.UnixConn) (int32, error) { + rawConnection, err := connection.SyscallConn() + if err != nil { + return 0, err + } + var credentials *unix.Ucred + var credentialErr error + if err := rawConnection.Control(func(fd uintptr) { + credentials, credentialErr = unix.GetsockoptUcred( + int(fd), unix.SOL_SOCKET, unix.SO_PEERCRED) + }); err != nil { + return 0, err + } + if credentialErr != nil { + return 0, credentialErr + } + if credentials == nil || credentials.Pid <= 0 { + return 0, errors.New("host PID broker received invalid peer credentials") + } + return credentials.Pid, nil +} + +func (broker *Broker) Close() error { + broker.closeOnce.Do(func() { + broker.handlerMu.Lock() + broker.closing.Store(true) + broker.handlerMu.Unlock() + + listenerErr := broker.listener.Close() + broker.handlerWG.Wait() + removeErr := broker.removeOwnedSocket() + unlockErr := unix.Flock(int(broker.lockFile.Fd()), unix.LOCK_UN) + lockCloseErr := broker.lockFile.Close() + broker.closeErr = errors.Join(listenerErr, removeErr, unlockErr, + lockCloseErr) + }) + return broker.closeErr +} + +func (broker *Broker) removeOwnedSocket() error { + var stat unix.Stat_t + if err := unix.Lstat(broker.socketPath, &stat); err != nil { + if errors.Is(err, os.ErrNotExist) { + return nil + } + return fmt.Errorf("inspect host PID broker socket during cleanup: %w", + err) + } + if uint64(stat.Dev) != broker.socket.device || + stat.Ino != broker.socket.inode || + stat.Mode&unix.S_IFMT != unix.S_IFSOCK { + return nil + } + if err := unix.Unlink(broker.socketPath); err != nil && + !errors.Is(err, os.ErrNotExist) { + return fmt.Errorf("remove host PID broker socket: %w", err) + } + return nil +} diff --git a/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_linux_test.go b/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_linux_test.go new file mode 100644 index 0000000000..5594ed100d --- /dev/null +++ b/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_linux_test.go @@ -0,0 +1,396 @@ +//go:build linux + +/* + * SPDX-License-Identifier: Apache-2.0 + * + * Copyright (c) 2026 The HAMi Authors. + */ + +package hostpid + +import ( + "encoding/binary" + "errors" + "fmt" + "io" + "net" + "os" + "os/exec" + "path/filepath" + "strconv" + "sync" + "syscall" + "testing" + "time" +) + +const subprocessHelperEnvironment = "HAMI_HOSTPID_BROKER_HELPER" + +func startTestBroker(t *testing.T) (*Broker, string) { + t.Helper() + directory := t.TempDir() + socketPath := filepath.Join(directory, "broker.sock") + broker, err := listen(socketPath, os.Geteuid()) + if err != nil { + t.Fatalf("listen: %v", err) + } + serveResult := make(chan error, 1) + go func() { + serveResult <- broker.Serve() + }() + t.Cleanup(func() { + if err := broker.Close(); err != nil { + t.Errorf("close broker: %v", err) + } + if err := <-serveResult; err != nil { + t.Errorf("serve broker: %v", err) + } + }) + return broker, socketPath +} + +func queryBroker(socketPath string) (uint16, uint32, error) { + connection, err := net.DialTimeout("unix", socketPath, time.Second) + if err != nil { + return 0, 0, err + } + defer connection.Close() + if err := connection.SetDeadline(time.Now().Add(time.Second)); err != nil { + return 0, 0, err + } + request := []byte{'H', 'P', 'I', 'D', 0, 1, 0, 1} + if _, err := connection.Write(request); err != nil { + return 0, 0, err + } + response := make([]byte, responseSize) + if _, err := io.ReadFull(connection, response); err != nil { + return 0, 0, err + } + if string(response[:4]) != "HPID" || + binary.BigEndian.Uint16(response[4:6]) != protocolVersion { + return 0, 0, errors.New("invalid broker response") + } + return binary.BigEndian.Uint16(response[6:8]), + binary.BigEndian.Uint32(response[8:12]), nil +} + +func TestBrokerReturnsPeerPID(t *testing.T) { + _, socketPath := startTestBroker(t) + status, pid, err := queryBroker(socketPath) + if err != nil { + t.Fatal(err) + } + if status != statusOK || pid != uint32(os.Getpid()) { + t.Fatalf("got status %d PID %d, want status 0 PID %d", + status, pid, os.Getpid()) + } +} + +func TestBrokerReturnsSubprocessPID(t *testing.T) { + _, socketPath := startTestBroker(t) + command := exec.Command(os.Args[0], + "-test.run=^TestBrokerSubprocessHelper$") + command.Env = append(os.Environ(), + subprocessHelperEnvironment+"="+socketPath) + output, err := command.CombinedOutput() + if err != nil { + t.Fatalf("helper failed: %v\n%s", err, output) + } +} + +func TestBrokerSubprocessHelper(t *testing.T) { + socketPath := os.Getenv(subprocessHelperEnvironment) + if socketPath == "" { + return + } + status, pid, err := queryBroker(socketPath) + if err != nil { + t.Fatal(err) + } + if status != statusOK || pid != uint32(os.Getpid()) { + t.Fatalf("got status %d PID %d, want status 0 PID %d", + status, pid, os.Getpid()) + } +} + +func TestBrokerRejectsInvalidRequest(t *testing.T) { + _, socketPath := startTestBroker(t) + connection, err := net.Dial("unix", socketPath) + if err != nil { + t.Fatal(err) + } + defer connection.Close() + if _, err := connection.Write( + []byte{'B', 'A', 'D', '!', 0, 1, 0, 1}); err != nil { + t.Fatal(err) + } + response := make([]byte, responseSize) + if _, err := io.ReadFull(connection, response); err != nil { + t.Fatal(err) + } + if status := binary.BigEndian.Uint16(response[6:8]); status != statusInvalidRequest { + t.Fatalf("got status %d", status) + } + if pid := binary.BigEndian.Uint32(response[8:12]); pid != 0 { + t.Fatalf("got PID %d", pid) + } +} + +func TestBrokerTimesOutPartialRequest(t *testing.T) { + _, socketPath := startTestBroker(t) + connection, err := net.Dial("unix", socketPath) + if err != nil { + t.Fatal(err) + } + if _, err := connection.Write([]byte{'H', 'P', 'I', 'D'}); err != nil { + t.Fatal(err) + } + time.Sleep(transactionTimeout + 100*time.Millisecond) + buffer := make([]byte, 1) + if count, err := connection.Read(buffer); count != 0 || err == nil { + t.Fatalf("partial request connection stayed open: n=%d err=%v", + count, err) + } + _ = connection.Close() + + status, pid, err := queryBroker(socketPath) + if err != nil || status != statusOK || pid != uint32(os.Getpid()) { + t.Fatalf("broker did not recover: status=%d pid=%d err=%v", + status, pid, err) + } +} + +func TestBrokerHandlesConcurrentClients(t *testing.T) { + _, socketPath := startTestBroker(t) + const clients = 300 + errorsChannel := make(chan error, clients) + var waitGroup sync.WaitGroup + + for range clients { + waitGroup.Add(1) + go func() { + defer waitGroup.Done() + status, pid, err := queryBroker(socketPath) + if err != nil { + errorsChannel <- err + return + } + if status != statusOK || pid != uint32(os.Getpid()) { + errorsChannel <- fmt.Errorf("status=%d pid=%d", status, pid) + } + }() + } + waitGroup.Wait() + close(errorsChannel) + for err := range errorsChannel { + t.Error(err) + } +} + +func TestBrokerCreatesTrustedModes(t *testing.T) { + _, socketPath := startTestBroker(t) + directoryInfo, err := os.Stat(filepath.Dir(socketPath)) + if err != nil { + t.Fatal(err) + } + socketInfo, err := os.Stat(socketPath) + if err != nil { + t.Fatal(err) + } + if got := directoryInfo.Mode().Perm(); got != serverDirectoryMode { + t.Fatalf("directory mode is %#o", got) + } + if got := socketInfo.Mode().Perm(); got != serverSocketMode { + t.Fatalf("socket mode is %#o", got) + } +} + +func TestBrokerRejectsPathCollision(t *testing.T) { + directory := t.TempDir() + socketPath := filepath.Join(directory, "broker.sock") + if err := os.WriteFile(socketPath, []byte("keep"), 0o600); err != nil { + t.Fatal(err) + } + if broker, err := listen(socketPath, os.Geteuid()); err == nil { + _ = broker.Close() + t.Fatal("broker accepted a regular file collision") + } + contents, err := os.ReadFile(socketPath) + if err != nil || string(contents) != "keep" { + t.Fatalf("collision was changed: contents=%q err=%v", contents, err) + } +} + +func TestBrokerRemovesStaleSocket(t *testing.T) { + directory := t.TempDir() + socketPath := filepath.Join(directory, "broker.sock") + listener, err := net.ListenUnix("unix", &net.UnixAddr{ + Name: socketPath, + Net: "unix", + }) + if err != nil { + t.Fatal(err) + } + listener.SetUnlinkOnClose(false) + if err := listener.Close(); err != nil { + t.Fatal(err) + } + + broker, err := listen(socketPath, os.Geteuid()) + if err != nil { + t.Fatal(err) + } + if err := broker.Close(); err != nil { + t.Fatal(err) + } +} + +func TestBrokerDoesNotRemoveActiveSocket(t *testing.T) { + directory := t.TempDir() + socketPath := filepath.Join(directory, "broker.sock") + listener, err := net.ListenUnix("unix", &net.UnixAddr{ + Name: socketPath, + Net: "unix", + }) + if err != nil { + t.Fatal(err) + } + defer listener.Close() + + broker, err := listen(socketPath, os.Geteuid()) + if err == nil { + _ = broker.Close() + t.Fatal("broker replaced an active socket") + } + if _, err := os.Lstat(socketPath); err != nil { + t.Fatalf("active socket was removed: %v", err) + } +} + +func TestBrokerRejectsSymlinkDirectory(t *testing.T) { + parent := t.TempDir() + target := filepath.Join(parent, "target") + link := filepath.Join(parent, "link") + if err := os.Mkdir(target, 0o700); err != nil { + t.Fatal(err) + } + if err := os.Symlink(target, link); err != nil { + t.Fatal(err) + } + broker, err := listen(filepath.Join(link, "broker.sock"), os.Geteuid()) + if err == nil { + _ = broker.Close() + t.Fatal("broker accepted a symlink directory") + } +} + +func TestBrokerRejectsSymlinkLock(t *testing.T) { + directory := t.TempDir() + target := filepath.Join(directory, "target") + lockPath := filepath.Join(directory, "broker.lock") + if err := os.WriteFile(target, []byte("keep"), 0o600); err != nil { + t.Fatal(err) + } + if err := os.Symlink(target, lockPath); err != nil { + t.Fatal(err) + } + broker, err := listen(filepath.Join(directory, "broker.sock"), + os.Geteuid()) + if err == nil { + _ = broker.Close() + t.Fatal("broker accepted a symlink lock") + } + contents, err := os.ReadFile(target) + if err != nil || string(contents) != "keep" { + t.Fatalf("lock target was changed: contents=%q err=%v", contents, err) + } +} + +func TestBrokerRejectsSecondListener(t *testing.T) { + broker, socketPath := startTestBroker(t) + second, err := listen(socketPath, os.Geteuid()) + if err == nil { + _ = second.Close() + t.Fatal("second broker acquired the socket") + } + status, pid, queryErr := queryBroker(socketPath) + if queryErr != nil || status != statusOK || pid != uint32(os.Getpid()) { + t.Fatalf("first broker stopped: status=%d pid=%d err=%v", + status, pid, queryErr) + } + _ = broker +} + +func TestBrokerLeavesReplacementDuringClose(t *testing.T) { + directory := t.TempDir() + socketPath := filepath.Join(directory, "broker.sock") + oldSocketPath := filepath.Join(directory, "old.sock") + broker, err := listen(socketPath, os.Geteuid()) + if err != nil { + t.Fatal(err) + } + serveResult := make(chan error, 1) + go func() { serveResult <- broker.Serve() }() + + if err := os.Rename(socketPath, oldSocketPath); err != nil { + t.Fatal(err) + } + if err := os.WriteFile(socketPath, []byte("replacement"), 0o600); err != nil { + t.Fatal(err) + } + if err := broker.Close(); err != nil { + t.Fatal(err) + } + if err := <-serveResult; err != nil { + t.Fatal(err) + } + contents, err := os.ReadFile(socketPath) + if err != nil || string(contents) != "replacement" { + t.Fatalf("replacement was changed: contents=%q err=%v", contents, err) + } + if err := os.Remove(socketPath); err != nil { + t.Fatal(err) + } + if err := os.Remove(oldSocketPath); err != nil { + t.Fatal(err) + } +} + +func TestEnabled(t *testing.T) { + tests := map[string]bool{ + "": false, + "0": false, + "1": true, + "true": true, + "false": true, + } + for value, expected := range tests { + t.Run(strconv.Quote(value), func(t *testing.T) { + if got := Enabled(value); got != expected { + t.Fatalf("Enabled(%q)=%v, want %v", value, got, expected) + } + }) + } +} + +func TestListenDefaultRequiresRoot(t *testing.T) { + if os.Geteuid() == 0 { + t.Skip("test requires a non-root process") + } + broker, err := ListenDefault() + if broker != nil || err == nil { + t.Fatalf("broker=%v err=%v", broker, err) + } +} + +func TestBrokerSocketIdentityUsesDeviceAndInode(t *testing.T) { + _, socketPath := startTestBroker(t) + info, err := os.Lstat(socketPath) + if err != nil { + t.Fatal(err) + } + stat, ok := info.Sys().(*syscall.Stat_t) + if !ok || stat.Ino == 0 { + t.Fatalf("invalid socket stat: %#v", info.Sys()) + } +} diff --git a/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_unsupported.go b/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_unsupported.go new file mode 100644 index 0000000000..b6694d8595 --- /dev/null +++ b/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_unsupported.go @@ -0,0 +1,27 @@ +//go:build !linux + +/* + * SPDX-License-Identifier: Apache-2.0 + * + * Copyright (c) 2026 The HAMi Authors. + */ + +package hostpid + +import "errors" + +var errUnsupported = errors.New("the host PID broker requires Linux") + +type Broker struct{} + +func ListenDefault() (*Broker, error) { + return nil, errUnsupported +} + +func (broker *Broker) Serve() error { + return errUnsupported +} + +func (broker *Broker) Close() error { + return nil +} diff --git a/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/config.go b/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/config.go new file mode 100644 index 0000000000..19dbdd498e --- /dev/null +++ b/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/config.go @@ -0,0 +1,21 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * Copyright (c) 2026 The HAMi Authors. + */ + +package hostpid + +const ( + EnvironmentVariable = "LIBVGPU_HOSTPID_BROKER" + + ServerDirectory = "/var/run/hami/hostpid" + ServerSocketPath = ServerDirectory + "/broker.sock" + + ContainerDirectory = "/tmp/vgpulock/hostpid" + ContainerSocketPath = ContainerDirectory + "/broker.sock" +) + +func Enabled(value string) bool { + return value != "" && value != "0" +} From da40a54dfc0cef032520cb66ced7806ba42e5e65 Mon Sep 17 00:00:00 2001 From: iemAnshuman Date: Tue, 4 Aug 2026 02:38:11 +0530 Subject: [PATCH 03/20] start the host pid broker with the plugin Signed-off-by: iemAnshuman --- cmd/device-plugin/nvidia/hostpid_broker.go | 47 +++++++++++++++++++ .../nvidia/hostpid_broker_test.go | 27 +++++++++++ cmd/device-plugin/nvidia/main.go | 38 +++++++++++++-- 3 files changed, 107 insertions(+), 5 deletions(-) create mode 100644 cmd/device-plugin/nvidia/hostpid_broker.go create mode 100644 cmd/device-plugin/nvidia/hostpid_broker_test.go diff --git a/cmd/device-plugin/nvidia/hostpid_broker.go b/cmd/device-plugin/nvidia/hostpid_broker.go new file mode 100644 index 0000000000..778c96f9e7 --- /dev/null +++ b/cmd/device-plugin/nvidia/hostpid_broker.go @@ -0,0 +1,47 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * Copyright (c) 2026 The HAMi Authors. + */ + +package main + +import ( + "os" + + "k8s.io/klog/v2" + + "github.com/Project-HAMi/HAMi/pkg/device-plugin/nvidiadevice/nvinternal/hostpid" +) + +type runningHostPIDBroker struct { + broker *hostpid.Broker + done chan struct{} + serveErr error +} + +func startHostPIDBroker() (*runningHostPIDBroker, error) { + if !hostpid.Enabled(os.Getenv(hostpid.EnvironmentVariable)) { + return nil, nil + } + broker, err := hostpid.ListenDefault() + if err != nil { + return nil, err + } + running := &runningHostPIDBroker{ + broker: broker, + done: make(chan struct{}), + } + go func() { + running.serveErr = broker.Serve() + close(running.done) + }() + klog.Infof("Host PID broker is listening on %s", hostpid.ServerSocketPath) + return running, nil +} + +func (running *runningHostPIDBroker) stop() error { + closeErr := running.broker.Close() + <-running.done + return closeErr +} diff --git a/cmd/device-plugin/nvidia/hostpid_broker_test.go b/cmd/device-plugin/nvidia/hostpid_broker_test.go new file mode 100644 index 0000000000..89a337a214 --- /dev/null +++ b/cmd/device-plugin/nvidia/hostpid_broker_test.go @@ -0,0 +1,27 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * Copyright (c) 2026 The HAMi Authors. + */ + +package main + +import ( + "testing" + + "github.com/Project-HAMi/HAMi/pkg/device-plugin/nvidiadevice/nvinternal/hostpid" +) + +func TestStartHostPIDBrokerDisabled(t *testing.T) { + t.Setenv(hostpid.EnvironmentVariable, "") + running, err := startHostPIDBroker() + if err != nil || running != nil { + t.Fatalf("running=%v err=%v", running, err) + } + + t.Setenv(hostpid.EnvironmentVariable, "0") + running, err = startHostPIDBroker() + if err != nil || running != nil { + t.Fatalf("running=%v err=%v", running, err) + } +} diff --git a/cmd/device-plugin/nvidia/main.go b/cmd/device-plugin/nvidia/main.go index c91454bf8a..adc42b302e 100644 --- a/cmd/device-plugin/nvidia/main.go +++ b/cmd/device-plugin/nvidia/main.go @@ -18,6 +18,7 @@ package main import ( "encoding/json" + "errors" "flag" "fmt" "os" @@ -259,7 +260,7 @@ func loadConfig(c *cli.Context, flags []cli.Flag) (*spec.Config, error) { return config, nil } -func start(c *cli.Context, o *options) error { +func start(c *cli.Context, o *options) (resultErr error) { util.NodeName = os.Getenv(util.NodeNameEnvName) client.InitGlobalClient() @@ -271,6 +272,23 @@ func start(c *cli.Context, o *options) error { } defer watcher.Close() + hostPIDBroker, err := startHostPIDBroker() + if err != nil { + return fmt.Errorf("failed to start host PID broker: %w", err) + } + var hostPIDBrokerDone <-chan struct{} + hostPIDBrokerFailureReported := false + if hostPIDBroker != nil { + hostPIDBrokerDone = hostPIDBroker.done + defer func() { + stopErr := hostPIDBroker.stop() + if !hostPIDBrokerFailureReported { + stopErr = errors.Join(stopErr, hostPIDBroker.serveErr) + } + resultErr = errors.Join(resultErr, stopErr) + }() + } + /*Loading config files*/ klog.Infof("Start working on node %s", util.NodeName) klog.Info("Starting OS watcher.") @@ -304,6 +322,16 @@ restart: // some messages, trigger a restart of the plugins, or exit the program. for { select { + case <-hostPIDBrokerDone: + hostPIDBrokerFailureReported = true + if hostPIDBroker.serveErr != nil { + resultErr = fmt.Errorf("host PID broker stopped: %w", + hostPIDBroker.serveErr) + } else { + resultErr = errors.New("host PID broker stopped unexpectedly") + } + goto exit + // If the restart timeout has expired, then restart the plugins case <-restartTimeout: goto restart @@ -336,11 +364,11 @@ restart: } } exit: - err = stopPlugins(plugins) - if err != nil { - return fmt.Errorf("error stopping plugins: %v", err) + if err := stopPlugins(plugins); err != nil { + resultErr = errors.Join(resultErr, + fmt.Errorf("error stopping plugins: %v", err)) } - return nil + return resultErr } func startPlugins(c *cli.Context, o *options) ([]plugin.Interface, bool, error) { From 1199ccf64a708de24b93fc9df7abc8ba43abd2d2 Mon Sep 17 00:00:00 2001 From: iemAnshuman Date: Tue, 4 Aug 2026 02:41:29 +0530 Subject: [PATCH 04/20] mount the host pid broker for workloads Signed-off-by: iemAnshuman --- .../nvinternal/plugin/hostpid_broker.go | 31 +++++++++++ .../nvinternal/plugin/hostpid_broker_test.go | 53 +++++++++++++++++++ .../nvidiadevice/nvinternal/plugin/server.go | 1 + 3 files changed, 85 insertions(+) create mode 100644 pkg/device-plugin/nvidiadevice/nvinternal/plugin/hostpid_broker.go create mode 100644 pkg/device-plugin/nvidiadevice/nvinternal/plugin/hostpid_broker_test.go diff --git a/pkg/device-plugin/nvidiadevice/nvinternal/plugin/hostpid_broker.go b/pkg/device-plugin/nvidiadevice/nvinternal/plugin/hostpid_broker.go new file mode 100644 index 0000000000..4af63dd576 --- /dev/null +++ b/pkg/device-plugin/nvidiadevice/nvinternal/plugin/hostpid_broker.go @@ -0,0 +1,31 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * Copyright (c) 2026 The HAMi Authors. + */ + +package plugin + +import ( + "os" + + "github.com/Project-HAMi/HAMi/pkg/device-plugin/nvidiadevice/nvinternal/hostpid" + kubeletdevicepluginv1beta1 "k8s.io/kubelet/pkg/apis/deviceplugin/v1beta1" +) + +func configureHostPIDBroker( + response *kubeletdevicepluginv1beta1.ContainerAllocateResponse) { + if !hostpid.Enabled(os.Getenv(hostpid.EnvironmentVariable)) { + return + } + if response.Envs == nil { + response.Envs = make(map[string]string) + } + response.Envs[hostpid.EnvironmentVariable] = "1" + response.Mounts = append(response.Mounts, + &kubeletdevicepluginv1beta1.Mount{ + ContainerPath: hostpid.ContainerDirectory, + HostPath: hostpid.ServerDirectory, + ReadOnly: true, + }) +} diff --git a/pkg/device-plugin/nvidiadevice/nvinternal/plugin/hostpid_broker_test.go b/pkg/device-plugin/nvidiadevice/nvinternal/plugin/hostpid_broker_test.go new file mode 100644 index 0000000000..dad6bb65d0 --- /dev/null +++ b/pkg/device-plugin/nvidiadevice/nvinternal/plugin/hostpid_broker_test.go @@ -0,0 +1,53 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * Copyright (c) 2026 The HAMi Authors. + */ + +package plugin + +import ( + "testing" + + "github.com/Project-HAMi/HAMi/pkg/device-plugin/nvidiadevice/nvinternal/hostpid" + "github.com/stretchr/testify/require" + kubeletdevicepluginv1beta1 "k8s.io/kubelet/pkg/apis/deviceplugin/v1beta1" +) + +func TestConfigureHostPIDBrokerDisabled(t *testing.T) { + for _, value := range []string{"", "0"} { + t.Run(value, func(t *testing.T) { + t.Setenv(hostpid.EnvironmentVariable, value) + response := &kubeletdevicepluginv1beta1.ContainerAllocateResponse{} + + configureHostPIDBroker(response) + + require.Empty(t, response.Envs) + require.Empty(t, response.Mounts) + }) + } +} + +func TestConfigureHostPIDBrokerEnabled(t *testing.T) { + t.Setenv(hostpid.EnvironmentVariable, "1") + existingMount := &kubeletdevicepluginv1beta1.Mount{ + ContainerPath: "/tmp/vgpulock", + HostPath: "/tmp/vgpulock", + } + response := &kubeletdevicepluginv1beta1.ContainerAllocateResponse{ + Envs: map[string]string{"KEEP": "yes"}, + Mounts: []*kubeletdevicepluginv1beta1.Mount{existingMount}, + } + + configureHostPIDBroker(response) + + require.Equal(t, "yes", response.Envs["KEEP"]) + require.Equal(t, "1", response.Envs[hostpid.EnvironmentVariable]) + require.Len(t, response.Mounts, 2) + require.Same(t, existingMount, response.Mounts[0]) + require.Equal(t, &kubeletdevicepluginv1beta1.Mount{ + ContainerPath: hostpid.ContainerDirectory, + HostPath: hostpid.ServerDirectory, + ReadOnly: true, + }, response.Mounts[1]) +} diff --git a/pkg/device-plugin/nvidiadevice/nvinternal/plugin/server.go b/pkg/device-plugin/nvidiadevice/nvinternal/plugin/server.go index 4640af7a71..d37a3a45cd 100644 --- a/pkg/device-plugin/nvidiadevice/nvinternal/plugin/server.go +++ b/pkg/device-plugin/nvidiadevice/nvinternal/plugin/server.go @@ -691,6 +691,7 @@ func (plugin *NvidiaDevicePlugin) Allocate(ctx context.Context, reqs *kubeletdev HostPath: "/tmp/vgpulock", ReadOnly: false}, ) + configureHostPIDBroker(response) found := false for _, val := range currentCtr.Env { if strings.Compare(val.Name, "CUDA_DISABLE_CONTROL") == 0 { From 3e6bc91d52c2a8c00cff0c0313dd5b451a1ab26e Mon Sep 17 00:00:00 2001 From: iemAnshuman Date: Tue, 4 Aug 2026 02:51:44 +0530 Subject: [PATCH 05/20] add the host pid broker chart gate Signed-off-by: iemAnshuman --- charts/hami/README.md | 2 ++ .../device-plugin/daemonsetnvidia.yaml | 17 +++++++++++++++++ charts/hami/values.yaml | 4 ++++ 3 files changed, 23 insertions(+) diff --git a/charts/hami/README.md b/charts/hami/README.md index 9df0a684f5..0a1bf656bd 100644 --- a/charts/hami/README.md +++ b/charts/hami/README.md @@ -192,6 +192,8 @@ This document provides detailed descriptions of all configurable values paramete |-----------|-------------|---------------| | `devicePlugin.pluginPath` | Plugin path | `/var/lib/kubelet/device-plugins` | | `devicePlugin.libPath` | Library path | `/usr/local/vgpu` | +| `devicePlugin.hostPID` | Use the host PID namespace for the device plugin | `true` | +| `devicePlugin.hostPIDBroker.enabled` | Let HAMi core ask the device plugin for its host PID. This requires `devicePlugin.hostPID` | `false` | | `devicePlugin.nvidiaNodeSelector` | NVIDIA node selector | `{"gpu": "on"}` | | `devicePlugin.updateStrategy.type` | Update strategy type | `RollingUpdate` | | `devicePlugin.updateStrategy.rollingUpdate.maxUnavailable` | Maximum unavailable count | `1` | diff --git a/charts/hami/templates/device-plugin/daemonsetnvidia.yaml b/charts/hami/templates/device-plugin/daemonsetnvidia.yaml index 0e785c363b..1f07c5654f 100644 --- a/charts/hami/templates/device-plugin/daemonsetnvidia.yaml +++ b/charts/hami/templates/device-plugin/daemonsetnvidia.yaml @@ -1,3 +1,6 @@ +{{- if and .Values.devicePlugin.enabled .Values.devicePlugin.hostPIDBroker.enabled (not .Values.devicePlugin.hostPID) }} +{{- fail "devicePlugin.hostPIDBroker requires devicePlugin.hostPID" }} +{{- end }} {{- if .Values.devicePlugin.enabled }} apiVersion: apps/v1 kind: DaemonSet @@ -94,6 +97,10 @@ spec: value: {{ .Values.devicePlugin.deviceListStrategy }} - name: HOOK_PATH value: {{ .Values.global.gpuHookPath }} + {{- if .Values.devicePlugin.hostPIDBroker.enabled }} + - name: LIBVGPU_HOSTPID_BROKER + value: "1" + {{- end }} {{- if typeIs "bool" .Values.devicePlugin.passDeviceSpecsEnabled }} - name: PASS_DEVICE_SPECS value: {{ .Values.devicePlugin.passDeviceSpecsEnabled | quote }} @@ -147,6 +154,10 @@ spec: subPath: device-config.yaml - name: cdi-root mountPath: /var/run/cdi + {{- if .Values.devicePlugin.hostPIDBroker.enabled }} + - name: hostpid-broker + mountPath: /var/run/hami/hostpid + {{- end }} {{- if typeIs "string" .Values.devicePlugin.nvidiaDriverRoot }} # We always mount the driver root at /driver-root in the container. # This is required for CDI detection to work correctly. @@ -239,6 +250,12 @@ spec: hostPath: path: /var/run/cdi type: DirectoryOrCreate + {{- if .Values.devicePlugin.hostPIDBroker.enabled }} + - name: hostpid-broker + hostPath: + path: /var/run/hami/hostpid + type: DirectoryOrCreate + {{- end }} - name: usrbin hostPath: path: /usr/bin diff --git a/charts/hami/values.yaml b/charts/hami/values.yaml index 0a25bc2665..a36743fba2 100644 --- a/charts/hami/values.yaml +++ b/charts/hami/values.yaml @@ -409,6 +409,10 @@ devicePlugin: podAnnotations: {} hostPID: true + # Let HAMi core ask the device plugin for its host PID. + # This requires hostPID to be true. + hostPIDBroker: + enabled: false hostNetwork: false securityContext: privileged: true From 6c1e2bd455814c4be703dde74362f74f2dea8972 Mon Sep 17 00:00:00 2001 From: iemAnshuman Date: Wed, 5 Aug 2026 05:52:43 +0530 Subject: [PATCH 06/20] harden the host PID broker integration Signed-off-by: iemAnshuman --- charts/hami/README.md | 2 +- .../nvidia/hostpid_broker_test.go | 18 +- docs/develop/hostpid-broker.md | 141 +++++++++++ .../nvinternal/hostpid/broker_linux.go | 10 +- .../nvinternal/hostpid/broker_linux_test.go | 218 ++++++++++++++++-- .../nvidiadevice/nvinternal/hostpid/config.go | 2 +- .../nvinternal/plugin/hostpid_broker.go | 7 + .../nvinternal/plugin/hostpid_broker_test.go | 36 ++- 8 files changed, 406 insertions(+), 28 deletions(-) create mode 100644 docs/develop/hostpid-broker.md diff --git a/charts/hami/README.md b/charts/hami/README.md index 0a1bf656bd..ff228c6fe6 100644 --- a/charts/hami/README.md +++ b/charts/hami/README.md @@ -193,7 +193,7 @@ This document provides detailed descriptions of all configurable values paramete | `devicePlugin.pluginPath` | Plugin path | `/var/lib/kubelet/device-plugins` | | `devicePlugin.libPath` | Library path | `/usr/local/vgpu` | | `devicePlugin.hostPID` | Use the host PID namespace for the device plugin | `true` | -| `devicePlugin.hostPIDBroker.enabled` | Let HAMi core ask the device plugin for its host PID. This requires `devicePlugin.hostPID` | `false` | +| `devicePlugin.hostPIDBroker.enabled` | Let HAMi core ask the device plugin for its host PID. This requires `devicePlugin.hostPID`. See [Host PID broker](../../docs/develop/hostpid-broker.md) | `false` | | `devicePlugin.nvidiaNodeSelector` | NVIDIA node selector | `{"gpu": "on"}` | | `devicePlugin.updateStrategy.type` | Update strategy type | `RollingUpdate` | | `devicePlugin.updateStrategy.rollingUpdate.maxUnavailable` | Maximum unavailable count | `1` | diff --git a/cmd/device-plugin/nvidia/hostpid_broker_test.go b/cmd/device-plugin/nvidia/hostpid_broker_test.go index 89a337a214..a5d83d49d3 100644 --- a/cmd/device-plugin/nvidia/hostpid_broker_test.go +++ b/cmd/device-plugin/nvidia/hostpid_broker_test.go @@ -13,15 +13,13 @@ import ( ) func TestStartHostPIDBrokerDisabled(t *testing.T) { - t.Setenv(hostpid.EnvironmentVariable, "") - running, err := startHostPIDBroker() - if err != nil || running != nil { - t.Fatalf("running=%v err=%v", running, err) - } - - t.Setenv(hostpid.EnvironmentVariable, "0") - running, err = startHostPIDBroker() - if err != nil || running != nil { - t.Fatalf("running=%v err=%v", running, err) + for _, value := range []string{"", "0", "true", "false", "01", " 1"} { + t.Run(value, func(t *testing.T) { + t.Setenv(hostpid.EnvironmentVariable, value) + running, err := startHostPIDBroker() + if err != nil || running != nil { + t.Fatalf("running=%v err=%v", running, err) + } + }) } } diff --git a/docs/develop/hostpid-broker.md b/docs/develop/hostpid-broker.md new file mode 100644 index 0000000000..9c82cdfb06 --- /dev/null +++ b/docs/develop/hostpid-broker.md @@ -0,0 +1,141 @@ +# Host PID broker + +Status: local draft. The feature is disabled by default. + +## Purpose + +HAMi-core needs the host PID of each CUDA process because NVML reports processes in the host PID namespace. The current fallback discovers that PID by creating a CUDA primary context while holding the post init lock. That work becomes serial when many processes call `cuInit()` together. + +The host PID broker returns the caller's own host PID from Linux `SO_PEERCRED`. It runs inside the NVIDIA device plugin, whose pod already uses the host PID namespace. It does not read host procfs and does not accept a PID supplied by the client. + +## Requirements + +1. The NVIDIA device plugin must run on Linux as root. + +2. `devicePlugin.hostPID` must remain `true`. + +3. The workload must use a HAMi-core build that supports protocol version 1 and the `LIBVGPU_HOSTPID_BROKER` gate. + +4. The container runtime must support the read only nested bind mount used for `/tmp/vgpulock/hostpid`. + +## Enablement + +Set the chart value below: + +```yaml +devicePlugin: + hostPID: true + hostPIDBroker: + enabled: true +``` + +The chart rejects a configuration that enables the broker while disabling the device plugin host PID namespace. + +When enabled, the chart does four things: + +1. It sets `LIBVGPU_HOSTPID_BROKER=1` in the device plugin. + +2. It mounts the host directory `/var/run/hami/hostpid` into the device plugin. + +3. The device plugin creates `/var/run/hami/hostpid/broker.sock` and serves protocol version 1. + +4. Each allocation that receives HAMi-core also receives `LIBVGPU_HOSTPID_BROKER=1` and a read only mount from `/var/run/hami/hostpid` to `/tmp/vgpulock/hostpid`. + +No value other than the exact string `1` enables the server or client. + +## Protocol + +The protocol uses one request and one response on a Unix stream connection. Every integer is unsigned and encoded in network byte order. + +| Field | Request bytes | Response bytes | +| --- | ---: | ---: | +| Magic `HPID` | 4 | 4 | +| Version | 2 | 2 | +| Command or status | 2 | 2 | +| Host PID | 0 | 4 | + +Protocol version 1 supports command 1, which means get the caller's host PID. Status 0 is success. Status 1 means the request was invalid. + +The server reads the peer credentials from the connected socket after validating the request. The client never sends a PID. + +## Security boundary + +1. The server requires effective UID 0 for its default path. + +2. The server directory is owned by root with mode `0711`. + +3. A root owned `0600` lock file prevents two brokers from replacing each other. + +4. The server rejects symlink directories, symlink lock files, regular file collisions, sockets owned by another UID, and active sockets. + +5. The server removes only a stale socket owned by the expected UID. During shutdown it removes the path only if its device and inode still match the socket it created. + +6. The workload sees the broker directory through a read only mount. The HAMi-core client checks the directory owner, directory write bits, socket type, socket owner, read only mount flag, and connected peer UID before trusting a response. + +7. A caller can request only its own PID. The kernel supplies that identity through `SO_PEERCRED` in the broker's host PID namespace. + +The socket is available only to workloads that receive the allocation mount. A workload can still create connection pressure. The server limits active handlers, applies one transaction deadline, and closes excess connections. A client that cannot complete the transaction uses the existing bounded HAMi-core fallback. + +## Failure behavior + +| Condition | Server behavior | HAMi-core behavior | +| --- | --- | --- | +| Feature disabled | No socket is created | Existing NVML discovery path | +| Server cannot start | Device plugin startup fails | No new allocation is served by that plugin instance | +| Socket missing or stale in a workload | No broker response | Existing NVML discovery path | +| Unsafe owner, mode, mount, or peer UID | Request is not trusted | Existing NVML discovery path | +| Malformed protocol reply | Request fails | Existing NVML discovery path | +| Slow or unresponsive broker | Server and client deadlines close the request | Existing NVML discovery path | +| Broker exits after plugin startup | Device plugin exits with the broker error | Kubernetes restarts the device plugin | + +The broker never returns a guessed PID. A successful reply contains the PID supplied by the kernel. Every other outcome is a failure that leaves PID discovery to the existing path. + +## Rollout + +1. Install the server capable HAMi release with `hostPIDBroker.enabled=false`. + +2. Install the compatible HAMi-core library. With no broker mount it uses the existing fallback. + +3. Enable `hostPIDBroker.enabled` and wait for every NVIDIA device plugin pod to become ready. + +4. Restart or recreate selected workloads so their allocation responses include the broker mount and environment gate. + +5. Validate the selected workloads before widening the rollout. Check correct host PID assignment, CUDA context accounting, broker use, and fallback behavior. + +Existing workloads do not gain a new mount when the device plugin changes. They continue through their existing path until they are recreated. + +## Rollback + +1. Set `hostPIDBroker.enabled=false`. + +2. Wait for the NVIDIA device plugin rollout. + +3. Recreate workloads when the broker mount should be removed. Workloads that still have a compatible broker mount can continue until they exit. + +4. A newer HAMi-core with no broker available uses the existing fallback, so the library does not need to be rolled back first. + +This rollout contract applies to the broker feature. The separate PR 248 lock migration still requires its own mixed binary policy. + +## Validation required before release + +1. Go race tests for the broker, lifecycle, and allocation integration. + +2. The actual C client to Go server contract test. + +3. Missing, stale, unsafe, malformed, slow, saturated, restarting, and dying broker cases. + +4. Linux and CUDA builds for HAMi-core plus the client and context accounting tests. + +5. Kubernetes validation with the feature disabled, enabled, rolled forward, and rolled back. + +6. Concurrent `cuInit()` and first primary context benchmarks with raw output, environment details, source revisions, and checksums. + +## Known limits + +1. The design depends on Linux `SO_PEERCRED` and a device plugin in the host PID namespace. + +2. Sandboxed runtimes must be tested in real execution. A mounted Unix socket may be blocked or may not preserve the peer identity needed by this design. + +3. Existing workloads need recreation to receive or remove the allocation mount. + +4. Additional NVIDIA architectures, driver versions, CRI-O, rootless runtimes, gVisor, and Kata remain separate compatibility cells until each is tested. diff --git a/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_linux.go b/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_linux.go index b284298b00..dd4312be25 100644 --- a/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_linux.go +++ b/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_linux.go @@ -58,6 +58,14 @@ func ListenDefault() (*Broker, error) { } func listen(socketPath string, ownerUID int) (*Broker, error) { + return listenWithHandlerLimit(socketPath, ownerUID, maxHandlers) +} + +func listenWithHandlerLimit(socketPath string, ownerUID int, + handlerLimit int) (*Broker, error) { + if handlerLimit <= 0 { + return nil, errors.New("host PID broker handler limit must be positive") + } directory := filepath.Dir(socketPath) if err := prepareDirectory(directory, ownerUID); err != nil { return nil, err @@ -107,7 +115,7 @@ func listen(socketPath string, ownerUID int) (*Broker, error) { socketPath: socketPath, socket: identity, lockFile: lockFile, - handlerSlots: make(chan struct{}, maxHandlers), + handlerSlots: make(chan struct{}, handlerLimit), }, nil } diff --git a/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_linux_test.go b/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_linux_test.go index 5594ed100d..a20146ef1f 100644 --- a/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_linux_test.go +++ b/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_linux_test.go @@ -18,6 +18,7 @@ import ( "os/exec" "path/filepath" "strconv" + "strings" "sync" "syscall" "testing" @@ -26,6 +27,10 @@ import ( const subprocessHelperEnvironment = "HAMI_HOSTPID_BROKER_HELPER" +const externalCClientEnvironment = "HAMI_HOSTPID_C_CLIENT" + +const externalCStormEnvironment = "HAMI_HOSTPID_C_STORM" + func startTestBroker(t *testing.T) (*Broker, string) { t.Helper() directory := t.TempDir() @@ -98,6 +103,36 @@ func TestBrokerReturnsSubprocessPID(t *testing.T) { } } +func TestBrokerExternalCClient(t *testing.T) { + clientPath := os.Getenv(externalCClientEnvironment) + if clientPath == "" { + t.Skip("HAMI_HOSTPID_C_CLIENT is not set") + } + _, socketPath := startTestBroker(t) + command := exec.Command(clientPath, socketPath) + output, err := command.CombinedOutput() + if err != nil { + t.Fatalf("external C client failed: %v\n%s", err, output) + } +} + +func TestBrokerExternalCClientStorm(t *testing.T) { + clientPath := os.Getenv(externalCStormEnvironment) + if clientPath == "" { + t.Skip("HAMI_HOSTPID_C_STORM is not set") + } + _, socketPath := startTestBroker(t) + command := exec.Command(clientPath, socketPath, "300") + output, err := command.CombinedOutput() + if err != nil { + t.Fatalf("external C client storm failed: %v\n%s", err, output) + } + if !strings.Contains(string(output), + "clients=300 successful=300 failed=0") { + t.Fatalf("unexpected C client storm output: %s", output) + } +} + func TestBrokerSubprocessHelper(t *testing.T) { socketPath := os.Getenv(subprocessHelperEnvironment) if socketPath == "" { @@ -114,25 +149,50 @@ func TestBrokerSubprocessHelper(t *testing.T) { } func TestBrokerRejectsInvalidRequest(t *testing.T) { + _, socketPath := startTestBroker(t) + tests := map[string][]byte{ + "magic": {'B', 'A', 'D', '!', 0, 1, 0, 1}, + "version": {'H', 'P', 'I', 'D', 0, 2, 0, 1}, + "command": {'H', 'P', 'I', 'D', 0, 1, 0, 2}, + } + for name, request := range tests { + t.Run(name, func(t *testing.T) { + connection, err := net.Dial("unix", socketPath) + if err != nil { + t.Fatal(err) + } + defer connection.Close() + if _, err := connection.Write(request); err != nil { + t.Fatal(err) + } + response := make([]byte, responseSize) + if _, err := io.ReadFull(connection, response); err != nil { + t.Fatal(err) + } + if status := binary.BigEndian.Uint16(response[6:8]); status != statusInvalidRequest { + t.Fatalf("got status %d", status) + } + if pid := binary.BigEndian.Uint32(response[8:12]); pid != 0 { + t.Fatalf("got PID %d", pid) + } + }) + } +} + +func TestBrokerRecoversFromEarlyClose(t *testing.T) { _, socketPath := startTestBroker(t) connection, err := net.Dial("unix", socketPath) if err != nil { t.Fatal(err) } - defer connection.Close() - if _, err := connection.Write( - []byte{'B', 'A', 'D', '!', 0, 1, 0, 1}); err != nil { + if err := connection.Close(); err != nil { t.Fatal(err) } - response := make([]byte, responseSize) - if _, err := io.ReadFull(connection, response); err != nil { - t.Fatal(err) - } - if status := binary.BigEndian.Uint16(response[6:8]); status != statusInvalidRequest { - t.Fatalf("got status %d", status) - } - if pid := binary.BigEndian.Uint32(response[8:12]); pid != 0 { - t.Fatalf("got PID %d", pid) + + status, pid, err := queryBroker(socketPath) + if err != nil || status != statusOK || pid != uint32(os.Getpid()) { + t.Fatalf("broker did not recover: status=%d pid=%d err=%v", + status, pid, err) } } @@ -187,6 +247,134 @@ func TestBrokerHandlesConcurrentClients(t *testing.T) { } } +func TestBrokerBoundsSlowClientsAndRecovers(t *testing.T) { + directory := t.TempDir() + socketPath := filepath.Join(directory, "broker.sock") + const handlerLimit = 4 + broker, err := listenWithHandlerLimit(socketPath, os.Geteuid(), + handlerLimit) + if err != nil { + t.Fatal(err) + } + serveResult := make(chan error, 1) + go func() { serveResult <- broker.Serve() }() + t.Cleanup(func() { + if err := broker.Close(); err != nil { + t.Errorf("close broker: %v", err) + } + if err := <-serveResult; err != nil { + t.Errorf("serve broker: %v", err) + } + }) + + connections := make([]net.Conn, 0, handlerLimit) + for range handlerLimit { + connection, err := net.Dial("unix", socketPath) + if err != nil { + t.Fatal(err) + } + connections = append(connections, connection) + } + deadline := time.Now().Add(time.Second) + for len(broker.handlerSlots) != handlerLimit && time.Now().Before(deadline) { + time.Sleep(time.Millisecond) + } + if got := len(broker.handlerSlots); got != handlerLimit { + t.Fatalf("active handlers=%d, want %d", got, handlerLimit) + } + + overflow, err := net.Dial("unix", socketPath) + if err != nil { + t.Fatal(err) + } + if err := overflow.SetDeadline(time.Now().Add(transactionTimeout)); err != nil { + t.Fatal(err) + } + request := []byte{'H', 'P', 'I', 'D', 0, 1, 0, 1} + _, _ = overflow.Write(request) + response := make([]byte, responseSize) + if _, err := io.ReadFull(overflow, response); err == nil { + t.Fatal("overflow client received a response") + } + _ = overflow.Close() + + for _, connection := range connections { + _ = connection.Close() + } + deadline = time.Now().Add(2 * transactionTimeout) + for { + status, pid, queryErr := queryBroker(socketPath) + if queryErr == nil && status == statusOK && pid == uint32(os.Getpid()) { + break + } + if time.Now().After(deadline) { + t.Fatalf("broker did not recover: status=%d pid=%d err=%v", + status, pid, queryErr) + } + time.Sleep(time.Millisecond) + } +} + +func TestBrokerCloseBoundsPartialClient(t *testing.T) { + directory := t.TempDir() + socketPath := filepath.Join(directory, "broker.sock") + broker, err := listenWithHandlerLimit(socketPath, os.Geteuid(), 1) + if err != nil { + t.Fatal(err) + } + serveResult := make(chan error, 1) + go func() { serveResult <- broker.Serve() }() + + connection, err := net.Dial("unix", socketPath) + if err != nil { + t.Fatal(err) + } + if _, err := connection.Write([]byte{'H'}); err != nil { + t.Fatal(err) + } + deadline := time.Now().Add(time.Second) + for len(broker.handlerSlots) != 1 && time.Now().Before(deadline) { + time.Sleep(time.Millisecond) + } + + begin := time.Now() + if err := broker.Close(); err != nil { + t.Fatal(err) + } + if elapsed := time.Since(begin); elapsed > 2*transactionTimeout { + t.Fatalf("close took %s", elapsed) + } + if err := <-serveResult; err != nil { + t.Fatal(err) + } + _ = connection.Close() +} + +func TestBrokerRestartsAfterClose(t *testing.T) { + directory := t.TempDir() + socketPath := filepath.Join(directory, "broker.sock") + + for attempt := range 2 { + broker, err := listen(socketPath, os.Geteuid()) + if err != nil { + t.Fatalf("listen attempt %d: %v", attempt, err) + } + serveResult := make(chan error, 1) + go func() { serveResult <- broker.Serve() }() + status, pid, err := queryBroker(socketPath) + if err != nil || status != statusOK || pid != uint32(os.Getpid()) { + t.Fatalf("query attempt %d: status=%d pid=%d err=%v", + attempt, status, pid, err) + } + if err := broker.Close(); err != nil { + t.Fatalf("close attempt %d: %v", attempt, err) + } + if err := <-serveResult; err != nil { + t.Fatalf("serve attempt %d: %v", attempt, err) + } + } +} + func TestBrokerCreatesTrustedModes(t *testing.T) { _, socketPath := startTestBroker(t) directoryInfo, err := os.Stat(filepath.Dir(socketPath)) @@ -361,8 +549,10 @@ func TestEnabled(t *testing.T) { "": false, "0": false, "1": true, - "true": true, - "false": true, + "true": false, + "false": false, + "01": false, + " 1": false, } for value, expected := range tests { t.Run(strconv.Quote(value), func(t *testing.T) { diff --git a/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/config.go b/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/config.go index 19dbdd498e..6421b59f33 100644 --- a/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/config.go +++ b/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/config.go @@ -17,5 +17,5 @@ const ( ) func Enabled(value string) bool { - return value != "" && value != "0" + return value == "1" } diff --git a/pkg/device-plugin/nvidiadevice/nvinternal/plugin/hostpid_broker.go b/pkg/device-plugin/nvidiadevice/nvinternal/plugin/hostpid_broker.go index 4af63dd576..d581b89f1e 100644 --- a/pkg/device-plugin/nvidiadevice/nvinternal/plugin/hostpid_broker.go +++ b/pkg/device-plugin/nvidiadevice/nvinternal/plugin/hostpid_broker.go @@ -22,6 +22,13 @@ func configureHostPIDBroker( response.Envs = make(map[string]string) } response.Envs[hostpid.EnvironmentVariable] = "1" + for _, mount := range response.Mounts { + if mount.ContainerPath == hostpid.ContainerDirectory { + mount.HostPath = hostpid.ServerDirectory + mount.ReadOnly = true + return + } + } response.Mounts = append(response.Mounts, &kubeletdevicepluginv1beta1.Mount{ ContainerPath: hostpid.ContainerDirectory, diff --git a/pkg/device-plugin/nvidiadevice/nvinternal/plugin/hostpid_broker_test.go b/pkg/device-plugin/nvidiadevice/nvinternal/plugin/hostpid_broker_test.go index dad6bb65d0..ffc54992a8 100644 --- a/pkg/device-plugin/nvidiadevice/nvinternal/plugin/hostpid_broker_test.go +++ b/pkg/device-plugin/nvidiadevice/nvinternal/plugin/hostpid_broker_test.go @@ -15,7 +15,7 @@ import ( ) func TestConfigureHostPIDBrokerDisabled(t *testing.T) { - for _, value := range []string{"", "0"} { + for _, value := range []string{"", "0", "true", "false", "01", " 1"} { t.Run(value, func(t *testing.T) { t.Setenv(hostpid.EnvironmentVariable, value) response := &kubeletdevicepluginv1beta1.ContainerAllocateResponse{} @@ -51,3 +51,37 @@ func TestConfigureHostPIDBrokerEnabled(t *testing.T) { ReadOnly: true, }, response.Mounts[1]) } + +func TestConfigureHostPIDBrokerIsIdempotent(t *testing.T) { + t.Setenv(hostpid.EnvironmentVariable, "1") + response := &kubeletdevicepluginv1beta1.ContainerAllocateResponse{} + + configureHostPIDBroker(response) + configureHostPIDBroker(response) + + require.Equal(t, "1", response.Envs[hostpid.EnvironmentVariable]) + require.Len(t, response.Mounts, 1) + require.Equal(t, &kubeletdevicepluginv1beta1.Mount{ + ContainerPath: hostpid.ContainerDirectory, + HostPath: hostpid.ServerDirectory, + ReadOnly: true, + }, response.Mounts[0]) +} + +func TestConfigureHostPIDBrokerReplacesConflictingMount(t *testing.T) { + t.Setenv(hostpid.EnvironmentVariable, "1") + response := &kubeletdevicepluginv1beta1.ContainerAllocateResponse{ + Mounts: []*kubeletdevicepluginv1beta1.Mount{{ + ContainerPath: hostpid.ContainerDirectory, + HostPath: "/untrusted", + ReadOnly: false, + }}, + } + + configureHostPIDBroker(response) + + require.Len(t, response.Mounts, 1) + require.Equal(t, hostpid.ServerDirectory, + response.Mounts[0].HostPath) + require.True(t, response.Mounts[0].ReadOnly) +} From a83c20fc0401722a5a716e880aefcab75cc637be Mon Sep 17 00:00:00 2001 From: iemAnshuman Date: Thu, 6 Aug 2026 17:19:45 +0530 Subject: [PATCH 07/20] test host PID broker allocation response Signed-off-by: iemAnshuman --- .../nvinternal/plugin/server_test.go | 21 +++++++++++++++++-- 1 file changed, 19 insertions(+), 2 deletions(-) diff --git a/pkg/device-plugin/nvidiadevice/nvinternal/plugin/server_test.go b/pkg/device-plugin/nvidiadevice/nvinternal/plugin/server_test.go index b69e7cafac..8bfec54cca 100644 --- a/pkg/device-plugin/nvidiadevice/nvinternal/plugin/server_test.go +++ b/pkg/device-plugin/nvidiadevice/nvinternal/plugin/server_test.go @@ -43,6 +43,7 @@ import ( v1 "github.com/NVIDIA/k8s-device-plugin/api/config/v1" "github.com/Project-HAMi/HAMi/pkg/device" "github.com/Project-HAMi/HAMi/pkg/device-plugin/nvidiadevice/nvinternal/cdi" + "github.com/Project-HAMi/HAMi/pkg/device-plugin/nvidiadevice/nvinternal/hostpid" "github.com/Project-HAMi/HAMi/pkg/device-plugin/nvidiadevice/nvinternal/imex" "github.com/Project-HAMi/HAMi/pkg/device-plugin/nvidiadevice/nvinternal/rm" "github.com/Project-HAMi/HAMi/pkg/device/nvidia" @@ -250,7 +251,7 @@ func TestCDIAllocateResponse(t *testing.T) { } for i := range testCases { - tc := testCases[i] + tc := &testCases[i] t.Run(tc.description, func(t *testing.T) { deviceListStrategies, _ := v1.NewDeviceListStrategies(tc.deviceListStrategies) plugin := NvidiaDevicePlugin{ @@ -963,7 +964,13 @@ func TestAlignContainerDevicesWithAllocatedIDsRejectsLengthMismatch(t *testing.T require.Contains(t, err.Error(), "device number not matched") } -func TestAllocateUsesKubeletSelectedUUIDsForVGPUResponse(t *testing.T) { +func TestAllocateUsesSelectedUUIDsAndHostPIDBroker(t *testing.T) { + t.Setenv(hostpid.EnvironmentVariable, "1") + previousEnableGetPreferredAllocation := enableGetPreferredAllocation + enableGetPreferredAllocation = true + defer func() { + enableGetPreferredAllocation = previousEnableGetPreferredAllocation + }() deviceListStrategies, _ := v1.NewDeviceListStrategies([]string{"envvar"}) deviceIDStrategy := v1.DeviceIDStrategyUUID memScale := 1.0 @@ -1033,6 +1040,16 @@ func TestAllocateUsesKubeletSelectedUUIDsForVGPUResponse(t *testing.T) { require.Equal(t, "GPU-03f69c50-207a-2038-9b45-23cac89cb67a", response.ContainerResponses[0].Envs[deviceListEnvVar]) require.Equal(t, "3000m", response.ContainerResponses[0].Envs["CUDA_DEVICE_MEMORY_LIMIT_0"]) require.Equal(t, "50", response.ContainerResponses[0].Envs["CUDA_DEVICE_SM_LIMIT"]) + require.Equal(t, "1", response.ContainerResponses[0].Envs[hostpid.EnvironmentVariable]) + brokerMountFound := false + for _, mount := range response.ContainerResponses[0].Mounts { + if mount.ContainerPath == hostpid.ContainerDirectory { + require.Equal(t, hostpid.ServerDirectory, mount.HostPath) + require.True(t, mount.ReadOnly) + brokerMountFound = true + } + } + require.True(t, brokerMountFound) } func TestAllocateReleasesNodeLockWhenNonMIGAllocateResponseFails(t *testing.T) { From de7de73969ac26dedbcb5b96dd36778a5a3b1d6a Mon Sep 17 00:00:00 2001 From: iemAnshuman Date: Sat, 8 Aug 2026 00:43:02 +0530 Subject: [PATCH 08/20] test: bound partial broker response reads Signed-off-by: iemAnshuman --- .../nvidiadevice/nvinternal/hostpid/broker_linux_test.go | 3 +++ 1 file changed, 3 insertions(+) diff --git a/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_linux_test.go b/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_linux_test.go index a20146ef1f..45318aced4 100644 --- a/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_linux_test.go +++ b/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_linux_test.go @@ -206,6 +206,9 @@ func TestBrokerTimesOutPartialRequest(t *testing.T) { t.Fatal(err) } time.Sleep(transactionTimeout + 100*time.Millisecond) + if err := connection.SetReadDeadline(time.Now().Add(time.Second)); err != nil { + t.Fatal(err) + } buffer := make([]byte, 1) if count, err := connection.Read(buffer); count != 0 || err == nil { t.Fatalf("partial request connection stayed open: n=%d err=%v", From 9d61b4196e9cc94c688eec82c9bf12afe6e19f78 Mon Sep 17 00:00:00 2001 From: iemAnshuman Date: Sat, 8 Aug 2026 00:43:25 +0530 Subject: [PATCH 09/20] test: verify recorded broker socket identity Signed-off-by: iemAnshuman --- .../nvidiadevice/nvinternal/hostpid/broker_linux_test.go | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_linux_test.go b/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_linux_test.go index 45318aced4..022b24bdb0 100644 --- a/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_linux_test.go +++ b/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_linux_test.go @@ -577,7 +577,7 @@ func TestListenDefaultRequiresRoot(t *testing.T) { } func TestBrokerSocketIdentityUsesDeviceAndInode(t *testing.T) { - _, socketPath := startTestBroker(t) + broker, socketPath := startTestBroker(t) info, err := os.Lstat(socketPath) if err != nil { t.Fatal(err) @@ -586,4 +586,9 @@ func TestBrokerSocketIdentityUsesDeviceAndInode(t *testing.T) { if !ok || stat.Ino == 0 { t.Fatalf("invalid socket stat: %#v", info.Sys()) } + if uint64(stat.Dev) != broker.socket.device || + stat.Ino != broker.socket.inode { + t.Fatalf("recorded identity dev=%d ino=%d, want dev=%d ino=%d", + broker.socket.device, broker.socket.inode, stat.Dev, stat.Ino) + } } From 4778c78193441abb80402ea12229d1ada778eb18 Mon Sep 17 00:00:00 2001 From: iemAnshuman Date: Sat, 8 Aug 2026 00:43:43 +0530 Subject: [PATCH 10/20] style: group host PID broker imports Signed-off-by: iemAnshuman --- .../nvidiadevice/nvinternal/plugin/hostpid_broker.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/pkg/device-plugin/nvidiadevice/nvinternal/plugin/hostpid_broker.go b/pkg/device-plugin/nvidiadevice/nvinternal/plugin/hostpid_broker.go index d581b89f1e..d9fa268037 100644 --- a/pkg/device-plugin/nvidiadevice/nvinternal/plugin/hostpid_broker.go +++ b/pkg/device-plugin/nvidiadevice/nvinternal/plugin/hostpid_broker.go @@ -9,8 +9,9 @@ package plugin import ( "os" - "github.com/Project-HAMi/HAMi/pkg/device-plugin/nvidiadevice/nvinternal/hostpid" kubeletdevicepluginv1beta1 "k8s.io/kubelet/pkg/apis/deviceplugin/v1beta1" + + "github.com/Project-HAMi/HAMi/pkg/device-plugin/nvidiadevice/nvinternal/hostpid" ) func configureHostPIDBroker( From 05a628c9a0bc8a0b2f34312980fb7df2229e0791 Mon Sep 17 00:00:00 2001 From: iemAnshuman Date: Sat, 8 Aug 2026 00:43:59 +0530 Subject: [PATCH 11/20] style: group host PID broker test imports Signed-off-by: iemAnshuman --- .../nvidiadevice/nvinternal/plugin/hostpid_broker_test.go | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/pkg/device-plugin/nvidiadevice/nvinternal/plugin/hostpid_broker_test.go b/pkg/device-plugin/nvidiadevice/nvinternal/plugin/hostpid_broker_test.go index ffc54992a8..14b0a55d5f 100644 --- a/pkg/device-plugin/nvidiadevice/nvinternal/plugin/hostpid_broker_test.go +++ b/pkg/device-plugin/nvidiadevice/nvinternal/plugin/hostpid_broker_test.go @@ -9,9 +9,10 @@ package plugin import ( "testing" - "github.com/Project-HAMi/HAMi/pkg/device-plugin/nvidiadevice/nvinternal/hostpid" "github.com/stretchr/testify/require" kubeletdevicepluginv1beta1 "k8s.io/kubelet/pkg/apis/deviceplugin/v1beta1" + + "github.com/Project-HAMi/HAMi/pkg/device-plugin/nvidiadevice/nvinternal/hostpid" ) func TestConfigureHostPIDBrokerDisabled(t *testing.T) { From 9a5b9421d8db475c9e382ffb4db67b93a5b0c5bb Mon Sep 17 00:00:00 2001 From: iemAnshuman Date: Sat, 8 Aug 2026 00:44:54 +0530 Subject: [PATCH 12/20] fix: retry transient broker accept errors Signed-off-by: iemAnshuman --- .../nvinternal/hostpid/broker_linux.go | 27 ++++++++++++++++ .../nvinternal/hostpid/broker_linux_test.go | 31 +++++++++++++++++++ 2 files changed, 58 insertions(+) diff --git a/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_linux.go b/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_linux.go index dd4312be25..71af146bbe 100644 --- a/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_linux.go +++ b/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_linux.go @@ -26,6 +26,8 @@ import ( const ( transactionTimeout = 500 * time.Millisecond activeProbeTimeout = 50 * time.Millisecond + acceptRetryInitial = 5 * time.Millisecond + acceptRetryMaximum = time.Second maxHandlers = 512 serverDirectoryMode = 0o711 serverSocketMode = 0o666 @@ -223,20 +225,45 @@ func readSocketIdentity(socketPath string, ownerUID int) (socketIdentity, error) } func (broker *Broker) Serve() error { + var backoff time.Duration for { connection, err := broker.listener.AcceptUnix() if err != nil { if broker.closing.Load() { return nil } + if isTemporaryAcceptError(err) { + backoff = nextAcceptBackoff(backoff) + time.Sleep(backoff) + continue + } return fmt.Errorf("accept host PID broker connection: %w", err) } + backoff = 0 if !broker.startHandler(connection) { _ = connection.Close() } } } +func isTemporaryAcceptError(err error) bool { + return errors.Is(err, syscall.EMFILE) || + errors.Is(err, syscall.ENFILE) || + errors.Is(err, syscall.ENOBUFS) || + errors.Is(err, syscall.ENOMEM) +} + +func nextAcceptBackoff(current time.Duration) time.Duration { + if current == 0 { + return acceptRetryInitial + } + next := current * 2 + if next > acceptRetryMaximum { + return acceptRetryMaximum + } + return next +} + func (broker *Broker) startHandler(connection *net.UnixConn) bool { broker.handlerMu.Lock() defer broker.handlerMu.Unlock() diff --git a/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_linux_test.go b/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_linux_test.go index 022b24bdb0..e9cdf62038 100644 --- a/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_linux_test.go +++ b/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_linux_test.go @@ -196,6 +196,37 @@ func TestBrokerRecoversFromEarlyClose(t *testing.T) { } } +func TestTemporaryAcceptErrors(t *testing.T) { + for _, acceptErr := range []error{ + syscall.EMFILE, + syscall.ENFILE, + syscall.ENOBUFS, + syscall.ENOMEM, + fmt.Errorf("wrapped: %w", syscall.EMFILE), + } { + if !isTemporaryAcceptError(acceptErr) { + t.Errorf("expected temporary accept error: %v", acceptErr) + } + } + if isTemporaryAcceptError(syscall.EINVAL) { + t.Fatal("EINVAL must remain a permanent accept error") + } +} + +func TestAcceptBackoffIsBounded(t *testing.T) { + backoff := time.Duration(0) + for range 32 { + backoff = nextAcceptBackoff(backoff) + if backoff > acceptRetryMaximum { + t.Fatalf("backoff %v exceeds maximum %v", backoff, + acceptRetryMaximum) + } + } + if backoff != acceptRetryMaximum { + t.Fatalf("backoff=%v, want %v", backoff, acceptRetryMaximum) + } +} + func TestBrokerTimesOutPartialRequest(t *testing.T) { _, socketPath := startTestBroker(t) connection, err := net.Dial("unix", socketPath) From 7b05d193463d4f41ceea988c60c45a9cdff2368d Mon Sep 17 00:00:00 2001 From: iemAnshuman Date: Sat, 8 Aug 2026 00:45:23 +0530 Subject: [PATCH 13/20] fix: rate limit broker transaction logs Signed-off-by: iemAnshuman --- .../nvinternal/hostpid/broker_linux.go | 19 ++++++++++++++++++- 1 file changed, 18 insertions(+), 1 deletion(-) diff --git a/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_linux.go b/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_linux.go index 71af146bbe..1b8e81b0da 100644 --- a/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_linux.go +++ b/pkg/device-plugin/nvidiadevice/nvinternal/hostpid/broker_linux.go @@ -21,6 +21,7 @@ import ( "time" "golang.org/x/sys/unix" + "k8s.io/klog/v2" ) const ( @@ -48,6 +49,7 @@ type Broker struct { handlerWG sync.WaitGroup handlerMu sync.Mutex closing atomic.Bool + dropped atomic.Uint64 closeOnce sync.Once closeErr error } @@ -290,6 +292,7 @@ func (broker *Broker) handle(connection *net.UnixConn) { request := make([]byte, requestSize) if _, err := io.ReadFull(connection, request); err != nil { + broker.logDroppedTransaction("request read", err) return } if !validRequest(request) { @@ -299,13 +302,27 @@ func (broker *Broker) handle(connection *net.UnixConn) { } pid, err := peerPID(connection) - if err != nil || pid <= 0 { + if err != nil { + broker.logDroppedTransaction("peer credentials", err) + return + } + if pid <= 0 { return } response := makeResponse(statusOK, uint32(pid)) writeResponse(connection, response) } +func (broker *Broker) logDroppedTransaction(operation string, err error) { + count := broker.dropped.Add(1) + if count&(count-1) != 0 { + return + } + klog.V(4).Infof( + "Dropped host PID broker transaction during %s (count=%d): %v", + operation, count, err) +} + func writeResponse(connection *net.UnixConn, response [responseSize]byte) { written := 0 From 9ad27a2ea8bf742c41eb0ea2abb1b310febaa8e8 Mon Sep 17 00:00:00 2001 From: iemAnshuman Date: Sat, 8 Aug 2026 00:46:58 +0530 Subject: [PATCH 14/20] fix: check broker health during plugin startup Signed-off-by: iemAnshuman --- cmd/device-plugin/nvidia/hostpid_broker.go | 17 ++++++++++++ .../nvidia/hostpid_broker_test.go | 26 +++++++++++++++++++ cmd/device-plugin/nvidia/main.go | 21 +++++++++------ 3 files changed, 56 insertions(+), 8 deletions(-) diff --git a/cmd/device-plugin/nvidia/hostpid_broker.go b/cmd/device-plugin/nvidia/hostpid_broker.go index 778c96f9e7..8fa181e7b0 100644 --- a/cmd/device-plugin/nvidia/hostpid_broker.go +++ b/cmd/device-plugin/nvidia/hostpid_broker.go @@ -7,6 +7,8 @@ package main import ( + "errors" + "fmt" "os" "k8s.io/klog/v2" @@ -45,3 +47,18 @@ func (running *runningHostPIDBroker) stop() error { <-running.done return closeErr } + +func (running *runningHostPIDBroker) failure() error { + if running == nil { + return nil + } + select { + case <-running.done: + if running.serveErr != nil { + return fmt.Errorf("host PID broker stopped: %w", running.serveErr) + } + return errors.New("host PID broker stopped unexpectedly") + default: + return nil + } +} diff --git a/cmd/device-plugin/nvidia/hostpid_broker_test.go b/cmd/device-plugin/nvidia/hostpid_broker_test.go index a5d83d49d3..3f77cd4d4e 100644 --- a/cmd/device-plugin/nvidia/hostpid_broker_test.go +++ b/cmd/device-plugin/nvidia/hostpid_broker_test.go @@ -7,6 +7,7 @@ package main import ( + "errors" "testing" "github.com/Project-HAMi/HAMi/pkg/device-plugin/nvidiadevice/nvinternal/hostpid" @@ -23,3 +24,28 @@ func TestStartHostPIDBrokerDisabled(t *testing.T) { }) } } + +func TestRunningHostPIDBrokerFailure(t *testing.T) { + var disabled *runningHostPIDBroker + if err := disabled.failure(); err != nil { + t.Fatalf("disabled broker failure=%v", err) + } + + running := &runningHostPIDBroker{done: make(chan struct{})} + if err := running.failure(); err != nil { + t.Fatalf("running broker failure=%v", err) + } + + serveErr := errors.New("accept failed") + running.serveErr = serveErr + close(running.done) + if err := running.failure(); !errors.Is(err, serveErr) { + t.Fatalf("stopped broker failure=%v, want %v", err, serveErr) + } + + stopped := &runningHostPIDBroker{done: make(chan struct{})} + close(stopped.done) + if err := stopped.failure(); err == nil { + t.Fatal("clean broker stop was not reported") + } +} diff --git a/cmd/device-plugin/nvidia/main.go b/cmd/device-plugin/nvidia/main.go index adc42b302e..44ce2051f6 100644 --- a/cmd/device-plugin/nvidia/main.go +++ b/cmd/device-plugin/nvidia/main.go @@ -307,7 +307,7 @@ restart: } klog.Info("Starting Plugins.") - plugins, restartPlugins, err := startPlugins(c, o) + plugins, restartPlugins, err := startPlugins(c, o, hostPIDBroker) if err != nil { return fmt.Errorf("error starting plugins: %v", err) } @@ -324,12 +324,7 @@ restart: select { case <-hostPIDBrokerDone: hostPIDBrokerFailureReported = true - if hostPIDBroker.serveErr != nil { - resultErr = fmt.Errorf("host PID broker stopped: %w", - hostPIDBroker.serveErr) - } else { - resultErr = errors.New("host PID broker stopped unexpectedly") - } + resultErr = hostPIDBroker.failure() goto exit // If the restart timeout has expired, then restart the plugins @@ -371,7 +366,8 @@ exit: return resultErr } -func startPlugins(c *cli.Context, o *options) ([]plugin.Interface, bool, error) { +func startPlugins(c *cli.Context, o *options, + hostPIDBroker *runningHostPIDBroker) ([]plugin.Interface, bool, error) { // Load the configuration file klog.Info("Loading configuration.") config, err := loadConfig(c, o.flags) @@ -430,17 +426,26 @@ func startPlugins(c *cli.Context, o *options) ([]plugin.Interface, bool, error) // to serve. If even one plugin fails to start properly, try // starting them all again. started := 0 + startedPlugins := make([]plugin.Interface, 0, len(plugins)) for _, p := range plugins { // Just continue if there are no devices to serve for plugin p. if len(p.Devices()) == 0 { continue } + if err := hostPIDBroker.failure(); err != nil { + return nil, false, errors.Join(err, stopPlugins(startedPlugins)) + } + // Start the gRPC server for plugin p and connect it with the kubelet. if err := p.Start(o.kubeletSocket); err != nil { klog.Errorf("Failed to start plugin: %v", err) return plugins, true, nil } + startedPlugins = append(startedPlugins, p) + if err := hostPIDBroker.failure(); err != nil { + return nil, false, errors.Join(err, stopPlugins(startedPlugins)) + } started++ } From 7c7ae67ddc3db63752b15add5780af72cee58dcb Mon Sep 17 00:00:00 2001 From: iemAnshuman Date: Sat, 8 Aug 2026 15:06:22 +0530 Subject: [PATCH 15/20] docs: update host PID broker status Signed-off-by: iemAnshuman --- docs/develop/hostpid-broker.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/develop/hostpid-broker.md b/docs/develop/hostpid-broker.md index 9c82cdfb06..5044b33b02 100644 --- a/docs/develop/hostpid-broker.md +++ b/docs/develop/hostpid-broker.md @@ -1,6 +1,6 @@ # Host PID broker -Status: local draft. The feature is disabled by default. +The feature is disabled by default. ## Purpose From 935a6dedfc23072c6f1b7ecc353dc5cc681b2591 Mon Sep 17 00:00:00 2001 From: iemAnshuman Date: Sat, 8 Aug 2026 15:11:47 +0530 Subject: [PATCH 16/20] test: cover host PID broker lifecycle Signed-off-by: iemAnshuman --- cmd/device-plugin/nvidia/hostpid_broker.go | 18 ++- .../nvidia/hostpid_broker_lifecycle_test.go | 130 ++++++++++++++++++ .../nvidia/hostpid_broker_test.go | 118 +++++++++++++++- cmd/device-plugin/nvidia/main.go | 38 +++-- 4 files changed, 288 insertions(+), 16 deletions(-) create mode 100644 cmd/device-plugin/nvidia/hostpid_broker_lifecycle_test.go diff --git a/cmd/device-plugin/nvidia/hostpid_broker.go b/cmd/device-plugin/nvidia/hostpid_broker.go index 8fa181e7b0..8f446aeecb 100644 --- a/cmd/device-plugin/nvidia/hostpid_broker.go +++ b/cmd/device-plugin/nvidia/hostpid_broker.go @@ -17,16 +17,30 @@ import ( ) type runningHostPIDBroker struct { - broker *hostpid.Broker + broker hostPIDBroker done chan struct{} serveErr error } +type hostPIDBroker interface { + Serve() error + Close() error +} + +type hostPIDBrokerListener func() (hostPIDBroker, error) + func startHostPIDBroker() (*runningHostPIDBroker, error) { + return startHostPIDBrokerWithListener(func() (hostPIDBroker, error) { + return hostpid.ListenDefault() + }) +} + +func startHostPIDBrokerWithListener( + listen hostPIDBrokerListener) (*runningHostPIDBroker, error) { if !hostpid.Enabled(os.Getenv(hostpid.EnvironmentVariable)) { return nil, nil } - broker, err := hostpid.ListenDefault() + broker, err := listen() if err != nil { return nil, err } diff --git a/cmd/device-plugin/nvidia/hostpid_broker_lifecycle_test.go b/cmd/device-plugin/nvidia/hostpid_broker_lifecycle_test.go new file mode 100644 index 0000000000..60b999d6c8 --- /dev/null +++ b/cmd/device-plugin/nvidia/hostpid_broker_lifecycle_test.go @@ -0,0 +1,130 @@ +/* + * SPDX-License-Identifier: Apache-2.0 + * + * Copyright (c) 2026 The HAMi Authors. + */ + +package main + +import ( + "errors" + "testing" + + "github.com/Project-HAMi/HAMi/pkg/device-plugin/nvidiadevice/nvinternal/plugin" + "github.com/Project-HAMi/HAMi/pkg/device-plugin/nvidiadevice/nvinternal/rm" +) + +type fakeDevicePlugin struct { + devices rm.Devices + start func(string) error + stopErr error + startCalls int + stopCalls int +} + +func (p *fakeDevicePlugin) Devices() rm.Devices { + return p.devices +} + +func (p *fakeDevicePlugin) Start(socket string) error { + p.startCalls++ + if p.start != nil { + return p.start(socket) + } + return nil +} + +func (p *fakeDevicePlugin) Stop() error { + p.stopCalls++ + return p.stopErr +} + +func devicePluginWithDevice() *fakeDevicePlugin { + return &fakeDevicePlugin{ + devices: rm.Devices{"GPU-0": &rm.Device{}}, + } +} + +func TestStartPluginServersDetectsBrokerFailureBeforeStart(t *testing.T) { + wantErr := errors.New("broker failed") + done := make(chan struct{}) + close(done) + running := &runningHostPIDBroker{ + done: done, + serveErr: wantErr, + } + p := devicePluginWithDevice() + + started, restart, err := startPluginServers( + []plugin.Interface{p}, "/tmp/kubelet.sock", running) + if started != 0 || restart || !errors.Is(err, wantErr) { + t.Fatalf("started=%d restart=%v err=%v, want broker failure", + started, restart, err) + } + if p.startCalls != 0 || p.stopCalls != 0 { + t.Fatalf("start calls=%d stop calls=%d, want 0 and 0", + p.startCalls, p.stopCalls) + } +} + +func TestStartPluginServersCleansUpAfterBrokerFailure(t *testing.T) { + wantServeErr := errors.New("broker failed") + wantStopErr := errors.New("plugin stop failed") + done := make(chan struct{}) + running := &runningHostPIDBroker{done: done} + p := devicePluginWithDevice() + p.stopErr = wantStopErr + p.start = func(string) error { + running.serveErr = wantServeErr + close(done) + return nil + } + + started, restart, err := startPluginServers( + []plugin.Interface{p}, "/tmp/kubelet.sock", running) + if started != 0 || restart || !errors.Is(err, wantServeErr) || + !errors.Is(err, wantStopErr) { + t.Fatalf("started=%d restart=%v err=%v, want joined failures", + started, restart, err) + } + if p.startCalls != 1 || p.stopCalls != 1 { + t.Fatalf("start calls=%d stop calls=%d, want 1 and 1", + p.startCalls, p.stopCalls) + } +} + +func TestStartPluginServersRequestsRestartAfterStartFailure(t *testing.T) { + wantErr := errors.New("plugin start failed") + p := devicePluginWithDevice() + p.start = func(socket string) error { + if socket != "/tmp/kubelet.sock" { + t.Fatalf("socket=%q", socket) + } + return wantErr + } + + started, restart, err := startPluginServers( + []plugin.Interface{p}, "/tmp/kubelet.sock", nil) + if started != 0 || !restart || err != nil { + t.Fatalf("started=%d restart=%v err=%v", started, restart, err) + } + if p.startCalls != 1 || p.stopCalls != 0 { + t.Fatalf("start calls=%d stop calls=%d, want 1 and 0", + p.startCalls, p.stopCalls) + } +} + +func TestStartPluginServersSkipsEmptyPlugins(t *testing.T) { + empty := &fakeDevicePlugin{} + ready := devicePluginWithDevice() + + started, restart, err := startPluginServers( + []plugin.Interface{empty, ready}, "/tmp/kubelet.sock", nil) + if started != 1 || restart || err != nil { + t.Fatalf("started=%d restart=%v err=%v", started, restart, err) + } + if empty.startCalls != 0 || ready.startCalls != 1 { + t.Fatalf("empty starts=%d ready starts=%d, want 0 and 1", + empty.startCalls, ready.startCalls) + } +} diff --git a/cmd/device-plugin/nvidia/hostpid_broker_test.go b/cmd/device-plugin/nvidia/hostpid_broker_test.go index 3f77cd4d4e..a425bc9282 100644 --- a/cmd/device-plugin/nvidia/hostpid_broker_test.go +++ b/cmd/device-plugin/nvidia/hostpid_broker_test.go @@ -8,20 +8,136 @@ package main import ( "errors" + "sync" "testing" "github.com/Project-HAMi/HAMi/pkg/device-plugin/nvidiadevice/nvinternal/hostpid" ) +type fakeHostPIDBroker struct { + serveStarted chan struct{} + serveRelease chan struct{} + closeOnce sync.Once + serveErr error + closeErr error +} + +func newFakeHostPIDBroker() *fakeHostPIDBroker { + return &fakeHostPIDBroker{ + serveStarted: make(chan struct{}), + serveRelease: make(chan struct{}), + } +} + +func (broker *fakeHostPIDBroker) Serve() error { + close(broker.serveStarted) + <-broker.serveRelease + return broker.serveErr +} + +func (broker *fakeHostPIDBroker) Close() error { + broker.closeOnce.Do(func() { + close(broker.serveRelease) + }) + return broker.closeErr +} + func TestStartHostPIDBrokerDisabled(t *testing.T) { for _, value := range []string{"", "0", "true", "false", "01", " 1"} { t.Run(value, func(t *testing.T) { t.Setenv(hostpid.EnvironmentVariable, value) - running, err := startHostPIDBroker() + listenerCalled := false + running, err := startHostPIDBrokerWithListener( + func() (hostPIDBroker, error) { + listenerCalled = true + return nil, nil + }) if err != nil || running != nil { t.Fatalf("running=%v err=%v", running, err) } + if listenerCalled { + t.Fatal("listener was called while broker was disabled") + } + }) + } +} + +func TestStartHostPIDBrokerDefaultDisabled(t *testing.T) { + t.Setenv(hostpid.EnvironmentVariable, "") + running, err := startHostPIDBroker() + if err != nil || running != nil { + t.Fatalf("running=%v err=%v", running, err) + } +} + +func TestStartHostPIDBrokerListenFailure(t *testing.T) { + t.Setenv(hostpid.EnvironmentVariable, "1") + wantErr := errors.New("listen failed") + running, err := startHostPIDBrokerWithListener( + func() (hostPIDBroker, error) { + return nil, wantErr + }) + if running != nil || !errors.Is(err, wantErr) { + t.Fatalf("running=%v err=%v, want %v", running, err, wantErr) + } +} + +func TestStartAndStopHostPIDBroker(t *testing.T) { + t.Setenv(hostpid.EnvironmentVariable, "1") + broker := newFakeHostPIDBroker() + running, err := startHostPIDBrokerWithListener( + func() (hostPIDBroker, error) { + return broker, nil + }) + if err != nil { + t.Fatal(err) + } + <-broker.serveStarted + if err := running.failure(); err != nil { + t.Fatalf("running broker failure=%v", err) + } + if err := running.stop(); err != nil { + t.Fatalf("stop broker: %v", err) + } + if err := running.failure(); err == nil { + t.Fatal("stopped broker was not reported") + } +} + +func TestHostPIDBrokerServeFailure(t *testing.T) { + t.Setenv(hostpid.EnvironmentVariable, "1") + wantErr := errors.New("serve failed") + broker := newFakeHostPIDBroker() + broker.serveErr = wantErr + close(broker.serveRelease) + running, err := startHostPIDBrokerWithListener( + func() (hostPIDBroker, error) { + return broker, nil }) + if err != nil { + t.Fatal(err) + } + <-running.done + if err := running.failure(); !errors.Is(err, wantErr) { + t.Fatalf("failure=%v, want %v", err, wantErr) + } +} + +func TestHostPIDBrokerCloseFailure(t *testing.T) { + t.Setenv(hostpid.EnvironmentVariable, "1") + wantErr := errors.New("close failed") + broker := newFakeHostPIDBroker() + broker.closeErr = wantErr + running, err := startHostPIDBrokerWithListener( + func() (hostPIDBroker, error) { + return broker, nil + }) + if err != nil { + t.Fatal(err) + } + <-broker.serveStarted + if err := running.stop(); !errors.Is(err, wantErr) { + t.Fatalf("stop=%v, want %v", err, wantErr) } } diff --git a/cmd/device-plugin/nvidia/main.go b/cmd/device-plugin/nvidia/main.go index 44ce2051f6..d61490098e 100644 --- a/cmd/device-plugin/nvidia/main.go +++ b/cmd/device-plugin/nvidia/main.go @@ -422,9 +422,24 @@ func startPlugins(c *cli.Context, o *options, return nil, false, fmt.Errorf("error getting plugins: %v", err) } - // Loop through all plugins, starting them if they have any devices - // to serve. If even one plugin fails to start properly, try - // starting them all again. + started, restartPlugins, err := startPluginServers(plugins, + o.kubeletSocket, hostPIDBroker) + if err != nil { + return nil, false, err + } + if restartPlugins { + return plugins, true, nil + } + + if started == 0 { + klog.Info("No devices found. Waiting indefinitely.") + } + + return plugins, false, nil +} + +func startPluginServers(plugins []plugin.Interface, kubeletSocket string, + hostPIDBroker *runningHostPIDBroker) (int, bool, error) { started := 0 startedPlugins := make([]plugin.Interface, 0, len(plugins)) for _, p := range plugins { @@ -434,26 +449,23 @@ func startPlugins(c *cli.Context, o *options, } if err := hostPIDBroker.failure(); err != nil { - return nil, false, errors.Join(err, stopPlugins(startedPlugins)) + return started, false, + errors.Join(err, stopPlugins(startedPlugins)) } // Start the gRPC server for plugin p and connect it with the kubelet. - if err := p.Start(o.kubeletSocket); err != nil { + if err := p.Start(kubeletSocket); err != nil { klog.Errorf("Failed to start plugin: %v", err) - return plugins, true, nil + return started, true, nil } startedPlugins = append(startedPlugins, p) if err := hostPIDBroker.failure(); err != nil { - return nil, false, errors.Join(err, stopPlugins(startedPlugins)) + return started, false, + errors.Join(err, stopPlugins(startedPlugins)) } started++ } - - if started == 0 { - klog.Info("No devices found. Waiting indefinitely.") - } - - return plugins, false, nil + return started, false, nil } func stopPlugins(plugins []plugin.Interface) error { From 2a2178e876a8844923646d8539663c92d83086aa Mon Sep 17 00:00:00 2001 From: iemAnshuman Date: Tue, 11 Aug 2026 18:39:51 +0530 Subject: [PATCH 17/20] fix: harden the host PID lock parent Prepare /tmp/vgpulock as a root owned sticky directory before allocation. Canonicalize the parent and broker mounts, keep the parent before the read only child, and remove stale broker settings when the gate is off. Reject allocation when the parent cannot be prepared safely. Signed-off-by: iemAnshuman --- docs/develop/hostpid-broker.md | 23 +- .../nvinternal/plugin/hostpid_broker.go | 224 +++++++++++- .../nvinternal/plugin/hostpid_broker_test.go | 330 ++++++++++++++++++ .../nvidiadevice/nvinternal/plugin/server.go | 11 +- .../nvinternal/plugin/server_test.go | 56 ++- 5 files changed, 620 insertions(+), 24 deletions(-) diff --git a/docs/develop/hostpid-broker.md b/docs/develop/hostpid-broker.md index 5044b33b02..cb178eec0a 100644 --- a/docs/develop/hostpid-broker.md +++ b/docs/develop/hostpid-broker.md @@ -18,6 +18,8 @@ The host PID broker returns the caller's own host PID from Linux `SO_PEERCRED`. 4. The container runtime must support the read only nested bind mount used for `/tmp/vgpulock/hostpid`. +5. The shared `/tmp/vgpulock` parent must be owned by root and use mode `01777`. The sticky bit allows legacy lock creation while preventing an ordinary workload user from replacing an entry owned by another user. + ## Enablement Set the chart value below: @@ -41,6 +43,14 @@ When enabled, the chart does four things: 4. Each allocation that receives HAMi-core also receives `LIBVGPU_HOSTPID_BROKER=1` and a read only mount from `/var/run/hami/hostpid` to `/tmp/vgpulock/hostpid`. +The device plugin prepares `/tmp/vgpulock` with mode `01777` before returning an allocation. This also applies when the broker is disabled and HAMi-core uses the existing fallback. Allocation fails if the directory cannot be prepared safely. + +Preparation opens `/tmp` without following a symlink, verifies its owner and sticky rule, then creates and opens `vgpulock` relative to that descriptor. The first `mkdirat()` requests mode `01777`. A descriptor based `chmod` restores bits removed by the process umask. A final identity and mode check rejects replacement during preparation. + +The allocation response contains one writable parent mount at `/tmp/vgpulock`. When the broker is enabled, it also contains one read only broker mount at `/tmp/vgpulock/hostpid`. The integration replaces duplicate or path equivalent entries with these canonical mounts and preserves unrelated mounts. It lists the parent before the nested broker mount so the parent does not hide the broker mount when the runtime applies the response. + +Before applying the current gate, the allocation helper clears the reserved broker environment key. When the broker is disabled, it also removes stale broker mounts while preserving the writable parent mount and unrelated mounts. + No value other than the exact string `1` enables the server or client. ## Protocol @@ -64,15 +74,17 @@ The server reads the peer credentials from the connected socket after validating 2. The server directory is owned by root with mode `0711`. -3. A root owned `0600` lock file prevents two brokers from replacing each other. +3. The shared lock parent is owned by root with mode `01777`. Allocation rejects an unsafe owner, object type, symlink, or final mode. + +4. A root owned `0600` lock file prevents two brokers from replacing each other. -4. The server rejects symlink directories, symlink lock files, regular file collisions, sockets owned by another UID, and active sockets. +5. The server rejects symlink directories, symlink lock files, regular file collisions, sockets owned by another UID, and active sockets. -5. The server removes only a stale socket owned by the expected UID. During shutdown it removes the path only if its device and inode still match the socket it created. +6. The server removes only a stale socket owned by the expected UID. During shutdown it removes the path only if its device and inode still match the socket it created. -6. The workload sees the broker directory through a read only mount. The HAMi-core client checks the directory owner, directory write bits, socket type, socket owner, read only mount flag, and connected peer UID before trusting a response. +7. The workload sees the broker directory through a read only mount. The HAMi-core client checks the directory owner, directory write bits, socket type, socket owner, read only mount flag, and connected peer UID before trusting a response. -7. A caller can request only its own PID. The kernel supplies that identity through `SO_PEERCRED` in the broker's host PID namespace. +8. A caller can request only its own PID. The kernel supplies that identity through `SO_PEERCRED` in the broker's host PID namespace. The socket is available only to workloads that receive the allocation mount. A workload can still create connection pressure. The server limits active handlers, applies one transaction deadline, and closes excess connections. A client that cannot complete the transaction uses the existing bounded HAMi-core fallback. @@ -82,6 +94,7 @@ The socket is available only to workloads that receive the allocation mount. A w | --- | --- | --- | | Feature disabled | No socket is created | Existing NVML discovery path | | Server cannot start | Device plugin startup fails | No new allocation is served by that plugin instance | +| Lock parent cannot be prepared safely | The allocation fails | No unsafe parent mount is returned | | Socket missing or stale in a workload | No broker response | Existing NVML discovery path | | Unsafe owner, mode, mount, or peer UID | Request is not trusted | Existing NVML discovery path | | Malformed protocol reply | Request fails | Existing NVML discovery path | diff --git a/pkg/device-plugin/nvidiadevice/nvinternal/plugin/hostpid_broker.go b/pkg/device-plugin/nvidiadevice/nvinternal/plugin/hostpid_broker.go index d9fa268037..5de2ba684f 100644 --- a/pkg/device-plugin/nvidiadevice/nvinternal/plugin/hostpid_broker.go +++ b/pkg/device-plugin/nvidiadevice/nvinternal/plugin/hostpid_broker.go @@ -7,33 +7,237 @@ package plugin import ( + "fmt" "os" + "path/filepath" + "syscall" + "golang.org/x/sys/unix" kubeletdevicepluginv1beta1 "k8s.io/kubelet/pkg/apis/deviceplugin/v1beta1" "github.com/Project-HAMi/HAMi/pkg/device-plugin/nvidiadevice/nvinternal/hostpid" ) +const hostPIDLockParentDirectory = "/tmp/vgpulock" + +const hostPIDLockParentMode = os.FileMode(0o777) | os.ModeSticky + +const hostPIDLockParentCreateMode = uint32(0o1777) + +const hostPIDLockTrustedOwner = uint32(0) + +var prepareHostPIDLockParentForAllocation = prepareDefaultHostPIDLockParent + +func prepareDefaultHostPIDLockParent() error { + return prepareHostPIDLockParent(hostPIDLockParentDirectory, + hostPIDLockTrustedOwner) +} + +func createHostPIDLockParent(parentFD int, baseName string) error { + return createHostPIDLockParentWith(unix.Mkdirat, parentFD, baseName) +} + +func createHostPIDLockParentWith( + mkdirat func(int, string, uint32) error, + parentFD int, baseName string) error { + err := mkdirat(parentFD, baseName, hostPIDLockParentCreateMode) + if err != nil && err != unix.EEXIST { + return err + } + return nil +} + +func prepareHostPIDLockParent(directory string, trustedOwner uint32) error { + cleanDirectory := filepath.Clean(directory) + if !filepath.IsAbs(cleanDirectory) || + cleanDirectory == string(filepath.Separator) { + return fmt.Errorf("directory must be an absolute non-root path") + } + parentDirectory := filepath.Dir(cleanDirectory) + baseName := filepath.Base(cleanDirectory) + parentFD, err := unix.Open(parentDirectory, + unix.O_RDONLY|unix.O_DIRECTORY|unix.O_CLOEXEC|unix.O_NOFOLLOW, 0) + if err != nil { + return fmt.Errorf("open parent directory without symlinks: %w", err) + } + parentFile := os.NewFile(uintptr(parentFD), parentDirectory) + if parentFile == nil { + _ = unix.Close(parentFD) + return fmt.Errorf("wrap parent directory descriptor") + } + defer parentFile.Close() + + parentInfo, err := parentFile.Stat() + if err != nil { + return fmt.Errorf("inspect parent directory: %w", err) + } + parentStat, ok := parentInfo.Sys().(*syscall.Stat_t) + parentMode := parentInfo.Mode() + if !ok || !parentInfo.IsDir() || parentStat.Uid != trustedOwner { + return fmt.Errorf("parent directory is not owned by trusted UID %d", + trustedOwner) + } + if parentMode.Perm()&0o022 != 0 && parentMode&os.ModeSticky == 0 { + return fmt.Errorf("writable parent directory is not sticky") + } + + if err := createHostPIDLockParent(parentFD, baseName); err != nil { + return fmt.Errorf("create directory: %w", err) + } + fd, err := unix.Openat(parentFD, baseName, + unix.O_RDONLY|unix.O_DIRECTORY|unix.O_CLOEXEC|unix.O_NOFOLLOW, 0) + if err != nil { + return fmt.Errorf("open directory without symlinks: %w", err) + } + file := os.NewFile(uintptr(fd), cleanDirectory) + if file == nil { + _ = unix.Close(fd) + return fmt.Errorf("wrap directory descriptor") + } + defer file.Close() + + openedInfo, err := file.Stat() + if err != nil { + return fmt.Errorf("inspect opened directory: %w", err) + } + openedStat, ok := openedInfo.Sys().(*syscall.Stat_t) + if !ok || !openedInfo.IsDir() || openedStat.Uid != trustedOwner { + return fmt.Errorf("directory is not owned by trusted UID %d", + trustedOwner) + } + if err := file.Chmod(hostPIDLockParentMode); err != nil { + return fmt.Errorf("set sticky directory mode: %w", err) + } + + verifiedInfo, err := file.Stat() + if err != nil { + return fmt.Errorf("verify opened directory: %w", err) + } + verifiedStat, ok := verifiedInfo.Sys().(*syscall.Stat_t) + var currentStat unix.Stat_t + if err := unix.Fstatat(parentFD, baseName, ¤tStat, + unix.AT_SYMLINK_NOFOLLOW); err != nil { + return fmt.Errorf("reinspect directory entry: %w", err) + } + if !ok || currentStat.Mode&unix.S_IFMT != unix.S_IFDIR || + currentStat.Dev != verifiedStat.Dev || + currentStat.Ino != verifiedStat.Ino || + currentStat.Uid != trustedOwner || + verifiedStat.Uid != trustedOwner || + currentStat.Mode&0o7777 != 0o1777 || + verifiedInfo.Mode()&(os.ModePerm|os.ModeSetuid|os.ModeSetgid| + os.ModeSticky) != + hostPIDLockParentMode { + return fmt.Errorf("directory changed while it was prepared") + } + return nil +} + func configureHostPIDBroker( response *kubeletdevicepluginv1beta1.ContainerAllocateResponse) { + if response.Envs != nil { + delete(response.Envs, hostpid.EnvironmentVariable) + } if !hostpid.Enabled(os.Getenv(hostpid.EnvironmentVariable)) { + removeHostPIDMounts(response, hostpid.ContainerDirectory) return } if response.Envs == nil { response.Envs = make(map[string]string) } response.Envs[hostpid.EnvironmentVariable] = "1" + configureCanonicalHostPIDMount(response, hostpid.ContainerDirectory, + hostpid.ServerDirectory, true) + orderHostPIDMountPair(response) +} + +func removeHostPIDMounts( + response *kubeletdevicepluginv1beta1.ContainerAllocateResponse, + containerPath string) { + configuredMounts := make([]*kubeletdevicepluginv1beta1.Mount, 0, + len(response.Mounts)) + for _, mount := range response.Mounts { + if mountTargetsContainerPath(mount, containerPath) { + continue + } + configuredMounts = append(configuredMounts, mount) + } + response.Mounts = configuredMounts +} + +func configureHostPIDLockParentMount( + response *kubeletdevicepluginv1beta1.ContainerAllocateResponse) { + configureCanonicalHostPIDMount(response, hostPIDLockParentDirectory, + hostPIDLockParentDirectory, false) +} + +func orderHostPIDMountPair( + response *kubeletdevicepluginv1beta1.ContainerAllocateResponse) { + parentIndex := -1 + brokerIndex := -1 + for index, mount := range response.Mounts { + if mount == nil { + continue + } + if parentIndex < 0 && + mountTargetsContainerPath(mount, hostPIDLockParentDirectory) { + parentIndex = index + } + if brokerIndex < 0 && + mountTargetsContainerPath(mount, hostpid.ContainerDirectory) { + brokerIndex = index + } + } + if parentIndex < 0 || brokerIndex < 0 || parentIndex < brokerIndex { + return + } + + parentMount := response.Mounts[parentIndex] + brokerMount := response.Mounts[brokerIndex] + orderedMounts := make([]*kubeletdevicepluginv1beta1.Mount, 0, + len(response.Mounts)) + for index, mount := range response.Mounts { + switch index { + case brokerIndex: + orderedMounts = append(orderedMounts, parentMount, brokerMount) + case parentIndex: + continue + default: + orderedMounts = append(orderedMounts, mount) + } + } + response.Mounts = orderedMounts +} + +func configureCanonicalHostPIDMount( + response *kubeletdevicepluginv1beta1.ContainerAllocateResponse, + containerPath string, hostPath string, readOnly bool) { + canonicalMount := &kubeletdevicepluginv1beta1.Mount{ + ContainerPath: containerPath, + HostPath: hostPath, + ReadOnly: readOnly, + } + canonicalMountAdded := false + configuredMounts := make([]*kubeletdevicepluginv1beta1.Mount, 0, + len(response.Mounts)+1) for _, mount := range response.Mounts { - if mount.ContainerPath == hostpid.ContainerDirectory { - mount.HostPath = hostpid.ServerDirectory - mount.ReadOnly = true - return + if mountTargetsContainerPath(mount, containerPath) { + if !canonicalMountAdded { + configuredMounts = append(configuredMounts, canonicalMount) + canonicalMountAdded = true + } + continue } + configuredMounts = append(configuredMounts, mount) } - response.Mounts = append(response.Mounts, - &kubeletdevicepluginv1beta1.Mount{ - ContainerPath: hostpid.ContainerDirectory, - HostPath: hostpid.ServerDirectory, - ReadOnly: true, - }) + if !canonicalMountAdded { + configuredMounts = append(configuredMounts, canonicalMount) + } + response.Mounts = configuredMounts +} + +func mountTargetsContainerPath( + mount *kubeletdevicepluginv1beta1.Mount, containerPath string) bool { + return mount != nil && filepath.Clean(filepath.Join( + string(filepath.Separator), mount.ContainerPath)) == containerPath } diff --git a/pkg/device-plugin/nvidiadevice/nvinternal/plugin/hostpid_broker_test.go b/pkg/device-plugin/nvidiadevice/nvinternal/plugin/hostpid_broker_test.go index 14b0a55d5f..dd04284376 100644 --- a/pkg/device-plugin/nvidiadevice/nvinternal/plugin/hostpid_broker_test.go +++ b/pkg/device-plugin/nvidiadevice/nvinternal/plugin/hostpid_broker_test.go @@ -7,14 +7,169 @@ package plugin import ( + "os" + "path/filepath" + "runtime" + "syscall" "testing" "github.com/stretchr/testify/require" + "golang.org/x/sys/unix" kubeletdevicepluginv1beta1 "k8s.io/kubelet/pkg/apis/deviceplugin/v1beta1" "github.com/Project-HAMi/HAMi/pkg/device-plugin/nvidiadevice/nvinternal/hostpid" ) +func TestMain(m *testing.M) { + prepareHostPIDLockParentForAllocation = func() error { return nil } + os.Exit(m.Run()) +} + +func TestPrepareHostPIDLockParent(t *testing.T) { + directory := filepath.Join(t.TempDir(), "vgpulock") + + require.Equal(t, uint32(0), hostPIDLockTrustedOwner) + require.NoError(t, os.MkdirAll(directory, 0o700)) + require.NoError(t, prepareHostPIDLockParent(directory, + uint32(os.Geteuid()))) + + info, err := os.Stat(directory) + require.NoError(t, err) + require.True(t, info.IsDir()) + require.Equal(t, hostPIDLockParentMode, + info.Mode()&(os.ModePerm|os.ModeSticky)) +} + +func TestPrepareHostPIDLockParentRequestsStickyCreateMode(t *testing.T) { + called := false + var requestedFD int + var requestedName string + var requestedMode uint32 + require.NoError(t, createHostPIDLockParentWith( + func(parentFD int, baseName string, mode uint32) error { + called = true + requestedFD = parentFD + requestedName = baseName + requestedMode = mode + return nil + }, 17, "vgpulock")) + require.True(t, called) + require.Equal(t, 17, requestedFD) + require.Equal(t, "vgpulock", requestedName) + require.Equal(t, uint32(0o1777), requestedMode) +} + +func TestPrepareHostPIDLockParentCreatesStickyModeOnLinux(t *testing.T) { + if runtime.GOOS != "linux" { + t.Skip("Linux mkdirat mode behavior is required") + } + + parent := t.TempDir() + parentFD, err := unix.Open(parent, + unix.O_RDONLY|unix.O_DIRECTORY|unix.O_CLOEXEC, 0) + require.NoError(t, err) + t.Cleanup(func() { + require.NoError(t, unix.Close(parentFD)) + }) + + var createErr error + func() { + oldUmask := unix.Umask(0) + defer unix.Umask(oldUmask) + createErr = createHostPIDLockParent(parentFD, "vgpulock") + }() + require.NoError(t, createErr) + + var createdStat unix.Stat_t + require.NoError(t, unix.Fstatat(parentFD, "vgpulock", &createdStat, + unix.AT_SYMLINK_NOFOLLOW)) + require.Equal(t, hostPIDLockParentCreateMode, + uint32(createdStat.Mode)&0o7777) +} + +func TestPrepareDefaultHostPIDLockParentAsRoot(t *testing.T) { + if os.Geteuid() != 0 || + os.Getenv("HAMI_TEST_PRODUCTION_PARENT") != "1" { + t.Skip("an isolated root mount namespace is required") + } + + require.NoError(t, prepareDefaultHostPIDLockParent()) + info, err := os.Stat(hostPIDLockParentDirectory) + require.NoError(t, err) + stat, ok := info.Sys().(*syscall.Stat_t) + require.True(t, ok) + require.Equal(t, hostPIDLockTrustedOwner, stat.Uid) + require.Equal(t, hostPIDLockParentMode, + info.Mode()&(os.ModePerm|os.ModeSticky)) +} + +func TestPrepareHostPIDLockParentRejectsUntrustedObjects(t *testing.T) { + t.Run("owner", func(t *testing.T) { + directory := filepath.Join(t.TempDir(), "vgpulock") + require.NoError(t, os.Mkdir(directory, 0o700)) + + require.Error(t, prepareHostPIDLockParent(directory, + uint32(os.Geteuid()+1))) + }) + + t.Run("regular file", func(t *testing.T) { + path := filepath.Join(t.TempDir(), "vgpulock") + require.NoError(t, os.WriteFile(path, nil, 0o600)) + + require.Error(t, prepareHostPIDLockParent(path, + uint32(os.Geteuid()))) + }) + + t.Run("symlink", func(t *testing.T) { + parent := t.TempDir() + target := filepath.Join(parent, "target") + link := filepath.Join(parent, "vgpulock") + require.NoError(t, os.Mkdir(target, 0o700)) + require.NoError(t, os.Symlink(target, link)) + + require.Error(t, prepareHostPIDLockParent(link, + uint32(os.Geteuid()))) + }) +} + +func TestPrepareHostPIDLockParentRejectsUntrustedParent(t *testing.T) { + t.Run("symlink", func(t *testing.T) { + fixture := t.TempDir() + realParent := filepath.Join(fixture, "real-parent") + linkParent := filepath.Join(fixture, "link-parent") + directory := filepath.Join(linkParent, "vgpulock") + realDirectory := filepath.Join(realParent, "vgpulock") + require.NoError(t, os.Mkdir(realParent, 0o700)) + require.NoError(t, os.Symlink(realParent, linkParent)) + + require.Error(t, prepareHostPIDLockParent(directory, + uint32(os.Geteuid()))) + _, err := os.Lstat(realDirectory) + require.ErrorIs(t, err, os.ErrNotExist) + }) + + t.Run("writable without sticky bit", func(t *testing.T) { + parent := filepath.Join(t.TempDir(), "parent") + directory := filepath.Join(parent, "vgpulock") + require.NoError(t, os.Mkdir(parent, 0o700)) + require.NoError(t, os.Chmod(parent, 0o777)) + + require.Error(t, prepareHostPIDLockParent(directory, + uint32(os.Geteuid()))) + }) +} + +func TestPrepareHostPIDLockParentAllowsStickyWritableParent(t *testing.T) { + parent := filepath.Join(t.TempDir(), "parent") + directory := filepath.Join(parent, "vgpulock") + require.NoError(t, os.Mkdir(parent, 0o700)) + require.NoError(t, os.Chmod(parent, + os.FileMode(0o777)|os.ModeSticky)) + + require.NoError(t, prepareHostPIDLockParent(directory, + uint32(os.Geteuid()))) +} + func TestConfigureHostPIDBrokerDisabled(t *testing.T) { for _, value := range []string{"", "0", "true", "false", "01", " 1"} { t.Run(value, func(t *testing.T) { @@ -29,6 +184,50 @@ func TestConfigureHostPIDBrokerDisabled(t *testing.T) { } } +func TestConfigureHostPIDBrokerDisabledRemovesStaleConfiguration(t *testing.T) { + t.Setenv(hostpid.EnvironmentVariable, "0") + before := &kubeletdevicepluginv1beta1.Mount{ + ContainerPath: "/before", + HostPath: "/before", + } + lockParent := &kubeletdevicepluginv1beta1.Mount{ + ContainerPath: hostPIDLockParentDirectory, + HostPath: hostPIDLockParentDirectory, + } + after := &kubeletdevicepluginv1beta1.Mount{ + ContainerPath: "/after", + HostPath: "/after", + } + response := &kubeletdevicepluginv1beta1.ContainerAllocateResponse{ + Envs: map[string]string{ + "KEEP": "yes", + hostpid.EnvironmentVariable: "1", + }, + Mounts: []*kubeletdevicepluginv1beta1.Mount{ + before, + { + ContainerPath: hostpid.ContainerDirectory + "/", + HostPath: "/first-stale", + }, + lockParent, + { + ContainerPath: "tmp/vgpulock/hostpid", + HostPath: "/second-stale", + }, + after, + }, + } + + configureHostPIDBroker(response) + + require.Equal(t, map[string]string{"KEEP": "yes"}, response.Envs) + require.Equal(t, []*kubeletdevicepluginv1beta1.Mount{ + before, + lockParent, + after, + }, response.Mounts) +} + func TestConfigureHostPIDBrokerEnabled(t *testing.T) { t.Setenv(hostpid.EnvironmentVariable, "1") existingMount := &kubeletdevicepluginv1beta1.Mount{ @@ -86,3 +285,134 @@ func TestConfigureHostPIDBrokerReplacesConflictingMount(t *testing.T) { response.Mounts[0].HostPath) require.True(t, response.Mounts[0].ReadOnly) } + +func TestConfigureHostPIDBrokerCanonicalizesDuplicateMounts(t *testing.T) { + t.Setenv(hostpid.EnvironmentVariable, "1") + before := &kubeletdevicepluginv1beta1.Mount{ + ContainerPath: "/before", + HostPath: "/before", + } + after := &kubeletdevicepluginv1beta1.Mount{ + ContainerPath: "/after", + HostPath: "/after", + } + response := &kubeletdevicepluginv1beta1.ContainerAllocateResponse{ + Mounts: []*kubeletdevicepluginv1beta1.Mount{ + before, + { + ContainerPath: hostpid.ContainerDirectory + "/", + HostPath: "/first-untrusted", + ReadOnly: false, + }, + after, + { + ContainerPath: hostPIDLockParentDirectory + + "/hostpid/../hostpid", + HostPath: "/second-untrusted", + ReadOnly: false, + }, + }, + } + + configureHostPIDBroker(response) + + require.Len(t, response.Mounts, 3) + require.Same(t, before, response.Mounts[0]) + require.Equal(t, &kubeletdevicepluginv1beta1.Mount{ + ContainerPath: hostpid.ContainerDirectory, + HostPath: hostpid.ServerDirectory, + ReadOnly: true, + }, response.Mounts[1]) + require.Same(t, after, response.Mounts[2]) +} + +func TestConfigureHostPIDBrokerOrdersParentBeforeNestedMount(t *testing.T) { + t.Setenv(hostpid.EnvironmentVariable, "1") + before := &kubeletdevicepluginv1beta1.Mount{ + ContainerPath: "/before", + HostPath: "/before", + } + middle := &kubeletdevicepluginv1beta1.Mount{ + ContainerPath: "/middle", + HostPath: "/middle", + } + after := &kubeletdevicepluginv1beta1.Mount{ + ContainerPath: "/after", + HostPath: "/after", + } + response := &kubeletdevicepluginv1beta1.ContainerAllocateResponse{ + Mounts: []*kubeletdevicepluginv1beta1.Mount{ + before, + { + ContainerPath: "tmp/vgpulock/hostpid", + HostPath: "/untrusted-broker", + ReadOnly: false, + }, + middle, + { + ContainerPath: "/tmp/./vgpulock", + HostPath: "/untrusted-parent", + ReadOnly: true, + }, + after, + }, + } + + configureHostPIDLockParentMount(response) + configureHostPIDBroker(response) + + require.Equal(t, []*kubeletdevicepluginv1beta1.Mount{ + before, + { + ContainerPath: hostPIDLockParentDirectory, + HostPath: hostPIDLockParentDirectory, + ReadOnly: false, + }, + { + ContainerPath: hostpid.ContainerDirectory, + HostPath: hostpid.ServerDirectory, + ReadOnly: true, + }, + middle, + after, + }, response.Mounts) +} + +func TestConfigureHostPIDLockParentMountCanonicalizesDuplicateMounts( + t *testing.T) { + before := &kubeletdevicepluginv1beta1.Mount{ + ContainerPath: "/before", + HostPath: "/before", + } + after := &kubeletdevicepluginv1beta1.Mount{ + ContainerPath: "/after", + HostPath: "/after", + } + response := &kubeletdevicepluginv1beta1.ContainerAllocateResponse{ + Mounts: []*kubeletdevicepluginv1beta1.Mount{ + before, + { + ContainerPath: hostPIDLockParentDirectory + "/", + HostPath: "/first-untrusted", + ReadOnly: true, + }, + after, + { + ContainerPath: "tmp/./vgpulock", + HostPath: "/second-untrusted", + ReadOnly: false, + }, + }, + } + + configureHostPIDLockParentMount(response) + + require.Len(t, response.Mounts, 3) + require.Same(t, before, response.Mounts[0]) + require.Equal(t, &kubeletdevicepluginv1beta1.Mount{ + ContainerPath: hostPIDLockParentDirectory, + HostPath: hostPIDLockParentDirectory, + ReadOnly: false, + }, response.Mounts[1]) + require.Same(t, after, response.Mounts[2]) +} diff --git a/pkg/device-plugin/nvidiadevice/nvinternal/plugin/server.go b/pkg/device-plugin/nvidiadevice/nvinternal/plugin/server.go index d37a3a45cd..3a1e1513d9 100644 --- a/pkg/device-plugin/nvidiadevice/nvinternal/plugin/server.go +++ b/pkg/device-plugin/nvidiadevice/nvinternal/plugin/server.go @@ -678,8 +678,11 @@ func (plugin *NvidiaDevicePlugin) Allocate(ctx context.Context, reqs *kubeletdev os.MkdirAll(cacheFileHostDirectory, 0777) os.Chmod(cacheFileHostDirectory, 0777) - os.MkdirAll("/tmp/vgpulock", 0777) - os.Chmod("/tmp/vgpulock", 0777) + if err := prepareHostPIDLockParentForAllocation(); err != nil { + PodAllocationFailed(nodename, current, NodeLockNvidia) + return nil, fmt.Errorf( + "failed to prepare host PID lock parent: %w", err) + } response.Mounts = append(response.Mounts, &kubeletdevicepluginv1beta1.Mount{ContainerPath: fmt.Sprintf("%s/vgpu/libvgpu.so", hostHookPath), HostPath: GetLibPath(), @@ -687,10 +690,8 @@ func (plugin *NvidiaDevicePlugin) Allocate(ctx context.Context, reqs *kubeletdev &kubeletdevicepluginv1beta1.Mount{ContainerPath: fmt.Sprintf("%s/vgpu", hostHookPath), HostPath: cacheFileHostDirectory, ReadOnly: false}, - &kubeletdevicepluginv1beta1.Mount{ContainerPath: "/tmp/vgpulock", - HostPath: "/tmp/vgpulock", - ReadOnly: false}, ) + configureHostPIDLockParentMount(response) configureHostPIDBroker(response) found := false for _, val := range currentCtr.Env { diff --git a/pkg/device-plugin/nvidiadevice/nvinternal/plugin/server_test.go b/pkg/device-plugin/nvidiadevice/nvinternal/plugin/server_test.go index 8bfec54cca..c9bb286c9a 100644 --- a/pkg/device-plugin/nvidiadevice/nvinternal/plugin/server_test.go +++ b/pkg/device-plugin/nvidiadevice/nvinternal/plugin/server_test.go @@ -35,6 +35,7 @@ package plugin import ( "context" "encoding/json" + "errors" "fmt" "os" "reflect" @@ -966,6 +967,16 @@ func TestAlignContainerDevicesWithAllocatedIDsRejectsLengthMismatch(t *testing.T func TestAllocateUsesSelectedUUIDsAndHostPIDBroker(t *testing.T) { t.Setenv(hostpid.EnvironmentVariable, "1") + prepareCalls := 0 + previousPrepareHostPIDLockParent := prepareHostPIDLockParentForAllocation + prepareHostPIDLockParentForAllocation = func() error { + prepareCalls++ + return nil + } + defer func() { + prepareHostPIDLockParentForAllocation = + previousPrepareHostPIDLockParent + }() previousEnableGetPreferredAllocation := enableGetPreferredAllocation enableGetPreferredAllocation = true defer func() { @@ -1037,19 +1048,56 @@ func TestAllocateUsesSelectedUUIDsAndHostPIDBroker(t *testing.T) { response, err := plugin.Allocate(context.Background(), request) require.NoError(t, err) + require.Equal(t, 1, prepareCalls) require.Equal(t, "GPU-03f69c50-207a-2038-9b45-23cac89cb67a", response.ContainerResponses[0].Envs[deviceListEnvVar]) require.Equal(t, "3000m", response.ContainerResponses[0].Envs["CUDA_DEVICE_MEMORY_LIMIT_0"]) require.Equal(t, "50", response.ContainerResponses[0].Envs["CUDA_DEVICE_SM_LIMIT"]) require.Equal(t, "1", response.ContainerResponses[0].Envs[hostpid.EnvironmentVariable]) - brokerMountFound := false - for _, mount := range response.ContainerResponses[0].Mounts { + brokerMountCount := 0 + brokerMountIndex := -1 + fallbackParentMountCount := 0 + fallbackParentMountIndex := -1 + for mountIndex, mount := range response.ContainerResponses[0].Mounts { if mount.ContainerPath == hostpid.ContainerDirectory { require.Equal(t, hostpid.ServerDirectory, mount.HostPath) require.True(t, mount.ReadOnly) - brokerMountFound = true + brokerMountIndex = mountIndex + brokerMountCount++ + } + if mount.ContainerPath == hostPIDLockParentDirectory { + require.Equal(t, hostPIDLockParentDirectory, mount.HostPath) + require.False(t, mount.ReadOnly) + fallbackParentMountIndex = mountIndex + fallbackParentMountCount++ } } - require.True(t, brokerMountFound) + require.Equal(t, 1, brokerMountCount) + require.Equal(t, 1, fallbackParentMountCount) + require.Less(t, fallbackParentMountIndex, brokerMountIndex) + + t.Setenv(hostpid.EnvironmentVariable, "") + disabledResponse, err := plugin.Allocate(context.Background(), request) + require.NoError(t, err) + require.Equal(t, 2, prepareCalls) + require.NotContains(t, disabledResponse.ContainerResponses[0].Envs, + hostpid.EnvironmentVariable) + fallbackMountCount := 0 + for _, mount := range disabledResponse.ContainerResponses[0].Mounts { + require.NotEqual(t, hostpid.ContainerDirectory, mount.ContainerPath) + if mount.ContainerPath == hostPIDLockParentDirectory { + require.Equal(t, hostPIDLockParentDirectory, mount.HostPath) + require.False(t, mount.ReadOnly) + fallbackMountCount++ + } + } + require.Equal(t, 1, fallbackMountCount) + + prepareHostPIDLockParentForAllocation = func() error { + return errors.New("parent preparation fixture") + } + failedResponse, err := plugin.Allocate(context.Background(), request) + require.Nil(t, failedResponse) + require.ErrorContains(t, err, "failed to prepare host PID lock parent") } func TestAllocateReleasesNodeLockWhenNonMIGAllocateResponseFails(t *testing.T) { From 24ead56392430a8a55843130497c01346a498c21 Mon Sep 17 00:00:00 2001 From: iemAnshuman Date: Tue, 11 Aug 2026 19:05:27 +0530 Subject: [PATCH 18/20] docs: fix compound modifiers Signed-off-by: iemAnshuman --- docs/develop/hostpid-broker.md | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/docs/develop/hostpid-broker.md b/docs/develop/hostpid-broker.md index cb178eec0a..6f300c7695 100644 --- a/docs/develop/hostpid-broker.md +++ b/docs/develop/hostpid-broker.md @@ -16,7 +16,7 @@ The host PID broker returns the caller's own host PID from Linux `SO_PEERCRED`. 3. The workload must use a HAMi-core build that supports protocol version 1 and the `LIBVGPU_HOSTPID_BROKER` gate. -4. The container runtime must support the read only nested bind mount used for `/tmp/vgpulock/hostpid`. +4. The container runtime must support the read-only nested bind mount used for `/tmp/vgpulock/hostpid`. 5. The shared `/tmp/vgpulock` parent must be owned by root and use mode `01777`. The sticky bit allows legacy lock creation while preventing an ordinary workload user from replacing an entry owned by another user. @@ -41,13 +41,13 @@ When enabled, the chart does four things: 3. The device plugin creates `/var/run/hami/hostpid/broker.sock` and serves protocol version 1. -4. Each allocation that receives HAMi-core also receives `LIBVGPU_HOSTPID_BROKER=1` and a read only mount from `/var/run/hami/hostpid` to `/tmp/vgpulock/hostpid`. +4. Each allocation that receives HAMi-core also receives `LIBVGPU_HOSTPID_BROKER=1` and a read-only mount from `/var/run/hami/hostpid` to `/tmp/vgpulock/hostpid`. The device plugin prepares `/tmp/vgpulock` with mode `01777` before returning an allocation. This also applies when the broker is disabled and HAMi-core uses the existing fallback. Allocation fails if the directory cannot be prepared safely. -Preparation opens `/tmp` without following a symlink, verifies its owner and sticky rule, then creates and opens `vgpulock` relative to that descriptor. The first `mkdirat()` requests mode `01777`. A descriptor based `chmod` restores bits removed by the process umask. A final identity and mode check rejects replacement during preparation. +Preparation opens `/tmp` without following a symlink, verifies its owner and sticky rule, then creates and opens `vgpulock` relative to that descriptor. The first `mkdirat()` requests mode `01777`. A descriptor-based `chmod` restores bits removed by the process umask. A final identity and mode check rejects replacement during preparation. -The allocation response contains one writable parent mount at `/tmp/vgpulock`. When the broker is enabled, it also contains one read only broker mount at `/tmp/vgpulock/hostpid`. The integration replaces duplicate or path equivalent entries with these canonical mounts and preserves unrelated mounts. It lists the parent before the nested broker mount so the parent does not hide the broker mount when the runtime applies the response. +The allocation response contains one writable parent mount at `/tmp/vgpulock`. When the broker is enabled, it also contains one read-only broker mount at `/tmp/vgpulock/hostpid`. The integration replaces duplicate or path-equivalent entries with these canonical mounts and preserves unrelated mounts. It lists the parent before the nested broker mount so the parent does not hide the broker mount when the runtime applies the response. Before applying the current gate, the allocation helper clears the reserved broker environment key. When the broker is disabled, it also removes stale broker mounts while preserving the writable parent mount and unrelated mounts. @@ -76,13 +76,13 @@ The server reads the peer credentials from the connected socket after validating 3. The shared lock parent is owned by root with mode `01777`. Allocation rejects an unsafe owner, object type, symlink, or final mode. -4. A root owned `0600` lock file prevents two brokers from replacing each other. +4. A root-owned `0600` lock file prevents two brokers from replacing each other. 5. The server rejects symlink directories, symlink lock files, regular file collisions, sockets owned by another UID, and active sockets. 6. The server removes only a stale socket owned by the expected UID. During shutdown it removes the path only if its device and inode still match the socket it created. -7. The workload sees the broker directory through a read only mount. The HAMi-core client checks the directory owner, directory write bits, socket type, socket owner, read only mount flag, and connected peer UID before trusting a response. +7. The workload sees the broker directory through a read-only mount. The HAMi-core client checks the directory owner, directory write bits, socket type, socket owner, read-only mount flag, and connected peer UID before trusting a response. 8. A caller can request only its own PID. The kernel supplies that identity through `SO_PEERCRED` in the broker's host PID namespace. From 5a4142813387a43fad022de7353d2e12e538c17a Mon Sep 17 00:00:00 2001 From: iemAnshuman Date: Tue, 11 Aug 2026 19:06:21 +0530 Subject: [PATCH 19/20] test: clarify parent ownership rejection Signed-off-by: iemAnshuman --- .../nvidiadevice/nvinternal/plugin/hostpid_broker_test.go | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/pkg/device-plugin/nvidiadevice/nvinternal/plugin/hostpid_broker_test.go b/pkg/device-plugin/nvidiadevice/nvinternal/plugin/hostpid_broker_test.go index dd04284376..0d7984aba6 100644 --- a/pkg/device-plugin/nvidiadevice/nvinternal/plugin/hostpid_broker_test.go +++ b/pkg/device-plugin/nvidiadevice/nvinternal/plugin/hostpid_broker_test.go @@ -104,12 +104,13 @@ func TestPrepareDefaultHostPIDLockParentAsRoot(t *testing.T) { } func TestPrepareHostPIDLockParentRejectsUntrustedObjects(t *testing.T) { - t.Run("owner", func(t *testing.T) { + t.Run("parent owner", func(t *testing.T) { directory := filepath.Join(t.TempDir(), "vgpulock") require.NoError(t, os.Mkdir(directory, 0o700)) - require.Error(t, prepareHostPIDLockParent(directory, - uint32(os.Geteuid()+1))) + err := prepareHostPIDLockParent(directory, uint32(os.Geteuid()+1)) + require.ErrorContains(t, err, + "parent directory is not owned by trusted UID") }) t.Run("regular file", func(t *testing.T) { From 146ee027b8b0ac44c8f8b3d48581f2cf69a485e9 Mon Sep 17 00:00:00 2001 From: iemAnshuman Date: Tue, 11 Aug 2026 19:07:34 +0530 Subject: [PATCH 20/20] test: reset allocation fixtures between paths Signed-off-by: iemAnshuman --- .../nvinternal/plugin/server_test.go | 17 +++++++++++++---- 1 file changed, 13 insertions(+), 4 deletions(-) diff --git a/pkg/device-plugin/nvidiadevice/nvinternal/plugin/server_test.go b/pkg/device-plugin/nvidiadevice/nvinternal/plugin/server_test.go index c9bb286c9a..87976a15c8 100644 --- a/pkg/device-plugin/nvidiadevice/nvinternal/plugin/server_test.go +++ b/pkg/device-plugin/nvidiadevice/nvinternal/plugin/server_test.go @@ -49,9 +49,11 @@ import ( "github.com/Project-HAMi/HAMi/pkg/device-plugin/nvidiadevice/nvinternal/rm" "github.com/Project-HAMi/HAMi/pkg/device/nvidia" "github.com/Project-HAMi/HAMi/pkg/util" + "github.com/Project-HAMi/HAMi/pkg/util/client" "github.com/stretchr/testify/require" corev1 "k8s.io/api/core/v1" metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/kubernetes/fake" kubeletdevicepluginv1beta1 "k8s.io/kubelet/pkg/apis/deviceplugin/v1beta1" ) @@ -1028,10 +1030,6 @@ func TestAllocateUsesSelectedUUIDsAndHostPIDBroker(t *testing.T) { getPendingPod = func(context.Context, string) (*corev1.Pod, error) { return pod, nil } defer func() { getPendingPod = previousGetPendingPod }() - previousEraseNextDeviceTypeFromAnnotation := eraseNextDeviceTypeFromAnnotation - eraseNextDeviceTypeFromAnnotation = func(string, corev1.Pod) error { return nil } - defer func() { eraseNextDeviceTypeFromAnnotation = previousEraseNextDeviceTypeFromAnnotation }() - previousPodAllocationFailed := podAllocationFailed podAllocationFailed = func(string, *corev1.Pod, string) {} defer func() { podAllocationFailed = previousPodAllocationFailed }() @@ -1040,6 +1038,11 @@ func TestAllocateUsesSelectedUUIDsAndHostPIDBroker(t *testing.T) { podAllocationTrySuccess = func(string, string, string, *corev1.Pod) {} defer func() { podAllocationTrySuccess = previousPodAllocationTrySuccess }() + // Provide a fake K8s client so the real patchErasedAnnotation can patch + previousKubeClient := client.KubeClient + client.KubeClient = fake.NewSimpleClientset(pod) + defer func() { client.KubeClient = previousKubeClient }() + request := &kubeletdevicepluginv1beta1.AllocateRequest{ ContainerRequests: []*kubeletdevicepluginv1beta1.ContainerAllocateRequest{{ DevicesIds: []string{"GPU-03f69c50-207a-2038-9b45-23cac89cb67a-0"}, @@ -1076,6 +1079,9 @@ func TestAllocateUsesSelectedUUIDsAndHostPIDBroker(t *testing.T) { require.Less(t, fallbackParentMountIndex, brokerMountIndex) t.Setenv(hostpid.EnvironmentVariable, "") + pod.Annotations["hami.io/vgpu-devices-to-allocate"] = + "GPU-annotated-a,NVIDIA,3000,50:;" + client.KubeClient = fake.NewSimpleClientset(pod) disabledResponse, err := plugin.Allocate(context.Background(), request) require.NoError(t, err) require.Equal(t, 2, prepareCalls) @@ -1095,6 +1101,9 @@ func TestAllocateUsesSelectedUUIDsAndHostPIDBroker(t *testing.T) { prepareHostPIDLockParentForAllocation = func() error { return errors.New("parent preparation fixture") } + pod.Annotations["hami.io/vgpu-devices-to-allocate"] = + "GPU-annotated-a,NVIDIA,3000,50:;" + client.KubeClient = fake.NewSimpleClientset(pod) failedResponse, err := plugin.Allocate(context.Background(), request) require.Nil(t, failedResponse) require.ErrorContains(t, err, "failed to prepare host PID lock parent")