DAOS-14707 control: Allow multiple agents simultaneously (#13465)

This allows for the case where multiple agents are used to allow
a client connections to multiple systems. The agent log prints a
NOTICE-level message to indicate the presence of other daos_agent
processes on startup.

Signed-off-by: Kris Jacque <kris.jacque@intel.com>

src/control/cmd/daos_agent/start.go

@@ -48,7 +48,7 @@ type startCmd struct {
 
 func (cmd *startCmd) Execute(_ []string) error {
 	if err := common.CheckDupeProcess(); err != nil {
-		return err
+		cmd.Notice(err.Error())
 	}
 
 	cmd.Infof("Starting %s (pid %d)", versionString(), os.Getpid())
@@ -123,8 +123,7 @@ func (cmd *startCmd) Execute(_ []string) error {
 	drpcSrvStart := time.Now()
 	err = drpcServer.Start(hwlocCtx)
 	if err != nil {
-		cmd.Errorf("Unable to start socket server on %s: %v", sockPath, err)
-		return err
+		return errors.Wrap(err, "unable to start dRPC server")
 	}
 	cmd.Debugf("dRPC socket server started: %s", time.Since(drpcSrvStart))
 

src/control/drpc/drpc_server.go

@@ -1,5 +1,5 @@
 //
-// (C) Copyright 2018-2022 Intel Corporation.
+// (C) Copyright 2018-2023 Intel Corporation.
 //
 // SPDX-License-Identifier: BSD-2-Clause-Patent
 //
@@ -87,28 +87,51 @@ func (d *DomainSocketServer) Listen(ctx context.Context) {
 
 // Start sets up the dRPC server socket and kicks off the listener goroutine.
 func (d *DomainSocketServer) Start(ctx context.Context) error {
-	// Just in case an old socket file is still lying around
-	if err := syscall.Unlink(d.sockFile); err != nil && !os.IsNotExist(err) {
-		return errors.Wrapf(err, "Unable to unlink %s", d.sockFile)
+	if d == nil {
+		return errors.New("DomainSocketServer is nil")
 	}
 
 	addr := &net.UnixAddr{Name: d.sockFile, Net: "unixpacket"}
+	if err := d.checkExistingSocket(ctx, addr); err != nil {
+		return err
+	}
+
 	lis, err := net.ListenUnix("unixpacket", addr)
 	if err != nil {
-		return errors.Wrapf(err, "Unable to listen on unix socket %s", d.sockFile)
+		return errors.Wrapf(err, "unable to listen on unix socket %s", d.sockFile)
 	}
 	d.listener = lis
 
-	// The only writer should be the I/O Engines which should be running as the same user as
-	// daos_server process.
 	if err := os.Chmod(d.sockFile, d.sockFileMode); err != nil {
-		return errors.Wrapf(err, "Unable to set permissions on %s", d.sockFile)
+		return errors.Wrapf(err, "unable to set permissions on %s", d.sockFile)
 	}
 
 	go d.Listen(ctx)
 	return nil
 }
 
+func (d *DomainSocketServer) checkExistingSocket(ctx context.Context, addr *net.UnixAddr) error {
+	conn, err := net.DialUnix("unixpacket", nil, addr)
+	if err == nil {
+		_ = conn.Close()
+		return FaultSocketFileInUse(d.sockFile)
+	}
+
+	if errors.Is(err, syscall.ENOENT) {
+		return nil
+	}
+
+	if errors.Is(err, syscall.ECONNREFUSED) {
+		// File exists but no one is listening - it's safe to delete.
+		if err := syscall.Unlink(addr.Name); err != nil && !os.IsNotExist(err) {
+			return errors.Wrap(err, "unlink old socket file")
+		}
+		return nil
+	}
+
+	return err
+}
+
 // RegisterRPCModule takes a Module and associates it with the given
 // DomainSocketServer so it can be used to process incoming dRPC calls.
 func (d *DomainSocketServer) RegisterRPCModule(mod Module) {

src/control/drpc/drpc_server_test.go

@@ -17,6 +17,8 @@ import (
 	"google.golang.org/protobuf/proto"
 
 	"github.com/daos-stack/daos/src/control/common/test"
+	"github.com/daos-stack/daos/src/control/fault"
+	"github.com/daos-stack/daos/src/control/fault/code"
 	"github.com/daos-stack/daos/src/control/logging"
 )
 
@@ -144,52 +146,119 @@ func TestNewDomainSocketServer(t *testing.T) {
 	test.AssertEqual(t, dss.sockFile, expectedSock, "wrong sockfile")
 }
 
-func TestServer_Start_CantUnlinkSocket(t *testing.T) {
-	log, buf := logging.NewTestLogger(t.Name())
-	defer test.ShowBufferOnFailure(t, buf)
-
-	tmpDir, tmpCleanup := test.CreateTestDir(t)
-	defer tmpCleanup()
-
-	path := filepath.Join(tmpDir, "test.sock")
-
-	// Forbid searching the directory
-	if err := os.Chmod(tmpDir, 0000); err != nil {
-		t.Fatalf("Couldn't change permissions on dir: %v", err)
+func TestDrpc_DomainSocketServer_Start(t *testing.T) {
+	sockPath := func(dir string) string {
+		return filepath.Join(dir, "test.sock")
 	}
-	defer func() {
-		_ = os.Chmod(tmpDir, 0700)
-	}()
-
-	dss, _ := NewDomainSocketServer(log, path, testFileMode)
-
-	err := dss.Start(test.Context(t))
-
-	test.CmpErr(t, errors.New("unlink"), err)
-}
 
-func TestServer_Start_CantListen(t *testing.T) {
-	log, buf := logging.NewTestLogger(t.Name())
-	defer test.ShowBufferOnFailure(t, buf)
-
-	tmpDir, tmpCleanup := test.CreateTestDir(t)
-	defer tmpCleanup()
-
-	path := filepath.Join(tmpDir, "test.sock")
-
-	// Forbid writing the directory
-	if err := os.Chmod(tmpDir, 0500); err != nil {
-		t.Fatalf("Couldn't change permissions on dir: %v", err)
+	for name, tc := range map[string]struct {
+		nilServer bool
+		setup     func(t *testing.T, dir string) func()
+		expErr    error
+	}{
+		"nil": {
+			nilServer: true,
+			expErr:    errors.New("nil"),
+		},
+		"unused existing socket file": {
+			setup: func(t *testing.T, dir string) func() {
+				t.Helper()
+
+				f, err := os.Create(sockPath(dir))
+				if err != nil {
+					t.Fatal(err)
+				}
+				_ = f.Close()
+				return func() {}
+			},
+		},
+		"can't unlink old socket file": {
+			setup: func(t *testing.T, dir string) func() {
+				t.Helper()
+
+				sockFile := sockPath(dir)
+				f, err := os.Create(sockFile)
+				if err != nil {
+					t.Fatal(err)
+				}
+				_ = f.Close()
+
+				if err := os.Chmod(dir, 0500); err != nil {
+					t.Fatalf("Couldn't change permissions on dir: %v", err)
+				}
+				return func() {
+					_ = os.Chmod(dir, 0700)
+				}
+			},
+			expErr: errors.New("unlink"),
+		},
+		"socket file in use": {
+			setup: func(t *testing.T, dir string) func() {
+				t.Helper()
+				log, buf := logging.NewTestLogger(t.Name())
+				defer test.ShowBufferOnFailure(t, buf)
+
+				other, err := NewDomainSocketServer(log, sockPath(dir), testFileMode)
+				if err != nil {
+					t.Fatalf("can't create first server: %s", err.Error())
+				}
+
+				err = other.Start(test.Context(t))
+				if err != nil {
+					t.Fatalf("can't start up first server: %s", err.Error())
+				}
+
+				// NB: The started server is shut down when the test context is canceled.
+				return func() {}
+			},
+			expErr: FaultSocketFileInUse(""),
+		},
+		"listen fails": {
+			setup: func(t *testing.T, dir string) func() {
+				t.Helper()
+
+				if err := os.Chmod(dir, 0500); err != nil {
+					t.Fatalf("Couldn't change permissions on dir: %v", err)
+				}
+				return func() {
+					_ = os.Chmod(dir, 0700)
+				}
+			},
+			expErr: errors.New("listen"),
+		},
+		"success": {},
+	} {
+		t.Run(name, func(t *testing.T) {
+			log, buf := logging.NewTestLogger(t.Name())
+			defer test.ShowBufferOnFailure(t, buf)
+
+			tmpDir, tmpCleanup := test.CreateTestDir(t)
+			defer tmpCleanup()
+
+			if tc.setup != nil {
+				teardown := tc.setup(t, tmpDir)
+				defer teardown()
+			}
+
+			// Test hack - make sure the right path is included in the fault message for comparison
+			if fault.IsFaultCode(tc.expErr, code.SocketFileInUse) {
+				tc.expErr = FaultSocketFileInUse(sockPath(tmpDir))
+			}
+
+			var err error
+			var dss *DomainSocketServer
+			if !tc.nilServer {
+				dss, err = NewDomainSocketServer(log, sockPath(tmpDir), testFileMode)
+				if err != nil {
+					t.Fatal(err)
+				}
+			}
+
+			err = dss.Start(test.Context(t))
+
+			test.CmpErr(t, tc.expErr, err)
+		})
 	}
-	defer func() {
-		_ = os.Chmod(tmpDir, 0700)
-	}()
-
-	dss, _ := NewDomainSocketServer(log, path, testFileMode)
-
-	err := dss.Start(test.Context(t))
-
-	test.CmpErr(t, errors.New("listen"), err)
 }
 
 func TestServer_RegisterModule(t *testing.T) {

src/control/drpc/fault.go

@@ -0,0 +1,27 @@
+//
+// (C) Copyright 2023 Intel Corporation.
+//
+// SPDX-License-Identifier: BSD-2-Clause-Patent
+//
+
+package drpc
+
+import (
+	"fmt"
+
+	"github.com/daos-stack/daos/src/control/fault"
+	"github.com/daos-stack/daos/src/control/fault/code"
+)
+
+// FaultSocketFileInUse indicates that the dRPC socket file was already in use when we tried
+// to start the dRPC server.
+func FaultSocketFileInUse(path string) *fault.Fault {
+	return &fault.Fault{
+		Domain:      "drpc",
+		Code:        code.SocketFileInUse,
+		Description: fmt.Sprintf("Configured dRPC socket file '%s' is already in use.", path),
+		Reason:      "dRPC socket file already in use",
+		Resolution: "If another process is using the socket file, configure a different socket directory. " +
+			"Otherwise, delete the existing socket file and try again.",
+	}
+}

src/control/fault/code/codes.go

@@ -50,6 +50,7 @@ const (
 	PrivilegedHelperNotPrivileged
 	PrivilegedHelperNotAvailable
 	PrivilegedHelperRequestFailed
+	SocketFileInUse
 )
 
 // generic storage fault codes

utils/test_memcheck.supp

@@ -383,3 +383,9 @@
    ...
    fun:mdb_txn_commit
 }
+{
+   go_runtime_syscall_param
+   Memcheck:Param
+   write(buf)
+   fun:runtime/internal/syscall.Syscall6
+}