nsqd: add Unix domain socket HTTP listener alongside TCP

apps/nsqd/options.go

@@ -131,6 +131,8 @@ func nsqdFlagSet(opts *nsqd.Options) *flag.FlagSet {
 	flagSet.String("https-address", opts.HTTPSAddress, "<addr>:<port> to listen on for HTTPS clients")
 	flagSet.String("http-address", opts.HTTPAddress, "address to listen on for HTTP clients (<addr>:<port> for TCP/IP or <path> for unix socket)")
 	flagSet.String("tcp-address", opts.TCPAddress, "address to listen on for TCP clients (<addr>:<port> for TCP/IP or <path> for unix socket)")
+	flagSet.String("unix-socket-path", opts.UnixSocketPath, "<path> to listen on for HTTP clients on a unix socket (in addition to --http-address)")
+	flagSet.String("unix-socket-permission", opts.UnixSocketPermission, "file permission of the unix socket as an octal number (e.g. 0660); empty means do not chmod")
 
 	authHTTPAddresses := app.StringArray{}
 	flagSet.Var(&authHTTPAddresses, "auth-http-address", "<addr>:<port> or a full url to query auth server (may be given multiple times)")

contrib/nsqd.cfg.example

@@ -13,6 +13,14 @@ http_address = "0.0.0.0:4151"
 ## <addr>:<port> to listen on for HTTPS clients
 # https_address = "0.0.0.0:4152"
 
+## <path> to listen on for HTTP clients on a unix socket
+## (in addition to http_address, served without TLS)
+# unix_socket_path = "/var/run/nsqd.sock"
+
+## file permission of the unix socket as an octal number (e.g. 0660);
+## empty/unset means do not chmod (umask applies)
+# unix_socket_permission = "0660"
+
 ## address that will be registered with lookupd (defaults to the OS hostname)
 # broadcast_address = ""
 

nsqd/nsqd.go

@@ -12,9 +12,11 @@ import (
 	"net"
 	"os"
 	"path"
+	"strconv"
 	"strings"
 	"sync"
 	"sync/atomic"
+	"syscall"
 	"time"
 
 	"github.com/nsqio/nsq/internal/clusterinfo"
@@ -61,6 +63,7 @@ type NSQD struct {
 	tcpListener     net.Listener
 	httpListener    net.Listener
 	httpsListener   net.Listener
+	udsListener     net.Listener
 	tlsConfig       *tls.Config
 	clientTLSConfig *tls.Config
 
@@ -165,6 +168,48 @@ func New(opts *Options) (*NSQD, error) {
 			return nil, fmt.Errorf("listen (%s) failed - %s", opts.HTTPSAddress, err)
 		}
 	}
+	if opts.UnixSocketPath != "" {
+		var mode os.FileMode
+		if opts.UnixSocketPermission != "" {
+			parsed, perr := strconv.ParseUint(opts.UnixSocketPermission, 8, 32)
+			if perr != nil {
+				return nil, fmt.Errorf("invalid --unix-socket-permission %q (expected octal, e.g. 0660) - %s", opts.UnixSocketPermission, perr)
+			}
+			mode = os.FileMode(parsed)
+		}
+		// Remove a leftover socket file from a previous crashed run so net.Listen
+		// does not fail with "address already in use". Only remove the path if it
+		// is actually a unix socket — never clobber a regular file or directory.
+		if info, statErr := os.Stat(opts.UnixSocketPath); statErr == nil {
+			if info.Mode()&os.ModeSocket == 0 {
+				return nil, fmt.Errorf("refusing to overwrite existing non-socket file at %s", opts.UnixSocketPath)
+			}
+			if rmErr := os.Remove(opts.UnixSocketPath); rmErr != nil {
+				return nil, fmt.Errorf("failed to remove stale unix socket %s - %s", opts.UnixSocketPath, rmErr)
+			}
+		}
+		// Narrow the umask while we bind so the socket inode is created with the
+		// intended permissions from the start, closing the race window between
+		// net.Listen and the follow-up os.Chmod.
+		if mode != 0 {
+			oldMask := syscall.Umask(int(^mode & 0777))
+			n.udsListener, err = net.Listen("unix", opts.UnixSocketPath)
+			syscall.Umask(oldMask)
+		} else {
+			n.udsListener, err = net.Listen("unix", opts.UnixSocketPath)
+		}
+		if err != nil {
+			return nil, fmt.Errorf("listen (%s) failed - %s", opts.UnixSocketPath, err)
+		}
+		if mode != 0 {
+			// Belt-and-braces: ensure the final mode matches even if the platform
+			// applied additional restrictions during Listen.
+			err = os.Chmod(opts.UnixSocketPath, mode)
+			if err != nil {
+				return nil, fmt.Errorf("failed to set unix socket permission - %s", err)
+			}
+		}
+	}
 	if opts.BroadcastHTTPPort == 0 {
 		tcpAddr, ok := n.RealHTTPAddr().(*net.TCPAddr)
 		if ok {
@@ -229,6 +274,13 @@ func (n *NSQD) RealHTTPSAddr() *net.TCPAddr {
 	return n.httpsListener.Addr().(*net.TCPAddr)
 }
 
+func (n *NSQD) RealUDSAddr() net.Addr {
+	if n.udsListener == nil {
+		return &net.UnixAddr{}
+	}
+	return n.udsListener.Addr()
+}
+
 func (n *NSQD) SetHealth(err error) {
 	n.errValue.Store(errStore{err: err})
 }
@@ -281,6 +333,15 @@ func (n *NSQD) Main() error {
 			exitFunc(http_api.Serve(n.httpsListener, httpsServer, "HTTPS", n.logf))
 		})
 	}
+	if n.udsListener != nil {
+		// The UDS HTTP listener is intentionally served as plain HTTP and
+		// ignores TLSRequired: access control is delegated to filesystem
+		// permissions on the socket path (see --unix-socket-permission).
+		udsHttpServer := newHTTPServer(n, false, false)
+		n.waitGroup.Wrap(func() {
+			exitFunc(http_api.Serve(n.udsListener, udsHttpServer, "HTTP", n.logf))
+		})
+	}
 
 	n.waitGroup.Wrap(n.queueScanLoop)
 	n.waitGroup.Wrap(n.lookupLoop)
@@ -460,6 +521,10 @@ func (n *NSQD) Exit() {
 		n.httpsListener.Close()
 	}
 
+	if n.udsListener != nil {
+		n.udsListener.Close()
+	}
+
 	n.Lock()
 	err := n.PersistMetadata()
 	if err != nil {

nsqd/options.go

@@ -22,6 +22,8 @@ type Options struct {
 	TCPAddress               string        `flag:"tcp-address"`
 	HTTPAddress              string        `flag:"http-address"`
 	HTTPSAddress             string        `flag:"https-address"`
+	UnixSocketPath           string        `flag:"unix-socket-path"`
+	UnixSocketPermission     string        `flag:"unix-socket-permission"`
 	BroadcastAddress         string        `flag:"broadcast-address"`
 	BroadcastTCPPort         int           `flag:"broadcast-tcp-port"`
 	BroadcastHTTPPort        int           `flag:"broadcast-http-port"`
@@ -105,6 +107,7 @@ func NewOptions() *Options {
 		TCPAddress:        "0.0.0.0:4150",
 		HTTPAddress:       "0.0.0.0:4151",
 		HTTPSAddress:      "0.0.0.0:4152",
+		UnixSocketPath:    "",
 		BroadcastAddress:  hostname,
 		BroadcastTCPPort:  0,
 		BroadcastHTTPPort: 0,