CSI: never replace a live socket; remove only our own socket on exit (#61)

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
 }
 

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

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)
+	}
+}