From 3dc6e5cb0a9656092e6800792ea7d11a5fb5bc7e Mon Sep 17 00:00:00 2001 From: gspark-etri Date: Thu, 22 Oct 2026 17:37:04 +0900 Subject: [PATCH] CSI: never replace a live socket; remove only our own socket on exit (#61) Co-Authored-By: Claude Opus 5.5 --- cmd/onggi/csi.go | 2 +- pkg/csi/server.go | 38 +++++++++++++++++++++++++++++++++++- pkg/csi/server_test.go | 44 ++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 82 insertions(+), 2 deletions(-) diff --git a/cmd/onggi/csi.go b/cmd/onggi/csi.go index bd83b9d..b95976d 100644 --- a/cmd/onggi/csi.go +++ b/cmd/onggi/csi.go @@ -143,7 +143,7 @@ func csiNode(args []string, stderr io.Writer) int { logger.Print(err) return 1 } - os.Remove(l.Addr().String()) + l.Close() // removes the socket if it is still ours (#61) return 0 } diff --git a/pkg/csi/server.go b/pkg/csi/server.go index c6146c1..8664405 100644 --- a/pkg/csi/server.go +++ b/pkg/csi/server.go @@ -10,6 +10,7 @@ import ( "path/filepath" "strings" "sync" + "time" csipb "github.com/container-storage-interface/spec/lib/go/csi" "google.golang.org/grpc" @@ -195,10 +196,45 @@ func Listen(endpoint string) (net.Listener, error) { if err := os.MkdirAll(filepath.Dir(p), 0o750); err != nil { return nil, err } + // a socket that answers belongs to a live plugin: do not take its place (#61) + if c, err := net.DialTimeout("unix", p, time.Second); err == nil { + c.Close() + return nil, fmt.Errorf("endpoint %s: another plugin is listening there", p) + } if err := os.Remove(p); err != nil && !errors.Is(err, os.ErrNotExist) { return nil, err } - return net.Listen("unix", p) + l, err := net.Listen("unix", p) + if err != nil { + return nil, err + } + ul := l.(*net.UnixListener) + ul.SetUnlinkOnClose(false) + fi, err := os.Stat(p) + if err != nil { + l.Close() + return nil, err + } + return &socket{UnixListener: ul, path: p, fi: fi}, nil +} + +// socket removes its path on Close only while the path is still this socket: a plugin that +// exits late must not remove the socket of the plugin that replaced it (#61). +type socket struct { + *net.UnixListener + path string + fi os.FileInfo + once sync.Once +} + +func (s *socket) Close() error { + err := s.UnixListener.Close() + s.once.Do(func() { + if fi, e := os.Stat(s.path); e == nil && os.SameFile(fi, s.fi) { + os.Remove(s.path) + } + }) + return err } // NewServer returns a gRPC server with n's Identity and Node services. Every failed call is diff --git a/pkg/csi/server_test.go b/pkg/csi/server_test.go index c031a12..189e689 100644 --- a/pkg/csi/server_test.go +++ b/pkg/csi/server_test.go @@ -3,6 +3,7 @@ package csi import ( "context" "net" + "os" "path/filepath" "strings" "sync" @@ -177,3 +178,46 @@ func TestServeOverUnixSocket(t *testing.T) { t.Fatalf("node info %v %v", ni, err) } } + +// A plugin does not take the place of one that answers, and a plugin that exits late leaves its +// successor's socket alone (#61). +func TestListenKeepsTheLiveSocket(t *testing.T) { + dir, err := os.MkdirTemp("", "csi") // short path: unix socket paths are limited to ~100 bytes + if err != nil { + t.Fatal(err) + } + defer os.RemoveAll(dir) + p := filepath.Join(dir, "csi.sock") + old, err := Listen("unix://" + p) + if err != nil { + t.Fatal(err) + } + go func() { + for { + c, err := old.Accept() + if err != nil { + return + } + c.Close() + } + }() + if l, err := Listen("unix://" + p); err == nil { + l.Close() + t.Fatal("took the place of a live plugin") + } + // the old plugin hangs; its socket file goes and a new plugin starts + os.Remove(p) + cur, err := Listen("unix://" + p) + if err != nil { + t.Fatal(err) + } + defer cur.Close() + old.Close() // exits late + if _, err := os.Stat(p); err != nil { + t.Fatalf("the late exit removed the live socket: %v", err) + } + cur.Close() + if _, err := os.Stat(p); !os.IsNotExist(err) { + t.Fatalf("a plugin left its own socket behind: %v", err) + } +}