refactor: support webserver with unix socket

Signed-off-by: thxCode <thxcode0824@gmail.com>

Author: thxCode

pkg/devicemanager/cmd.go

@@ -17,7 +17,7 @@ func NewCommand() *cobra.Command {
 	c := &cobra.Command{
 		Use: "device-manager",
 		Aliases: []string{
-			"devmgr",
+			"dm",
 		},
 		Short: "assists in managing hardware resources.",
 	}

pkg/devicemanager/option.go

@@ -49,7 +49,7 @@ func NewOptions() *Options {
 
 func (o *Options) AddFlags(fs *pflag.FlagSet) {
 	// Server.
-	o.ServerOptions.AddFlags(fs)
+	o.ServerOptions.AddFlags(fs, webserver.WithoutBindUnixPath())
 
 	// Manager.
 	o.ManagerOptions.AddFlags(fs, manager.WithoutKubeElectionOptions())

pkg/manager/manager.go

@@ -58,15 +58,15 @@ func (m *Manager) Prepare(_ context.Context) error {
 // before starting the controller manager.
 func (m *Manager) Start(ctx context.Context) error {
 	cm := m.CtrlManager
-	ms := cm.GetWebhookServer()
+	mu := cm.GetWebhookServer()
 
 	// Register /metrics.
 	{
 		h := promhttp.HandlerOpts{
 			ErrorLog:      klog.NewStandardLogger("WARNING"),
 			ErrorHandling: promhttp.HTTPErrorOnError,
 		}
-		ms.Register("/metrics", promhttp.HandlerFor(ctrlmetrics.Registry, h))
+		mu.Register("/metrics", promhttp.HandlerFor(ctrlmetrics.Registry, h))
 	}
 
 	// Register /readyz.
@@ -78,7 +78,7 @@ func (m *Manager) Start(ctx context.Context) error {
 				"log":  healthz.LogHealthz.Check,
 			},
 		}
-		ms.Register(p, http.StripPrefix(p, h))
+		mu.Register(p, http.StripPrefix(p, h))
 	}
 
 	// Register /livez.
@@ -102,18 +102,18 @@ func (m *Manager) Start(ctx context.Context) error {
 				},
 			},
 		}
-		ms.Register(p, http.StripPrefix(p, h))
+		mu.Register(p, http.StripPrefix(p, h))
 	}
 
 	// Register /debug.
 	{
 		runtime.SetBlockProfileRate(1)
-		ms.Register("/debug/pprof/", httpx.LoopbackAccessHandlerFunc(pprof.Index))
-		ms.Register("/debug/pprof/cmdline", httpx.LoopbackAccessHandlerFunc(pprof.Cmdline))
-		ms.Register("/debug/pprof/profile", httpx.LoopbackAccessHandlerFunc(pprof.Profile))
-		ms.Register("/debug/pprof/symbol", httpx.LoopbackAccessHandlerFunc(pprof.Symbol))
-		ms.Register("/debug/pprof/trace", httpx.LoopbackAccessHandlerFunc(pprof.Trace))
-		ms.Register("/debug/flags/v", httpx.LoopbackAccessHandlerFunc(routes.StringFlagPutHandler(logs.GlogSetter)))
+		mu.Register("/debug/pprof/", httpx.LoopbackAccessHandlerFunc(pprof.Index))
+		mu.Register("/debug/pprof/cmdline", httpx.LoopbackAccessHandlerFunc(pprof.Cmdline))
+		mu.Register("/debug/pprof/profile", httpx.LoopbackAccessHandlerFunc(pprof.Profile))
+		mu.Register("/debug/pprof/symbol", httpx.LoopbackAccessHandlerFunc(pprof.Symbol))
+		mu.Register("/debug/pprof/trace", httpx.LoopbackAccessHandlerFunc(pprof.Trace))
+		mu.Register("/debug/flags/v", httpx.LoopbackAccessHandlerFunc(routes.StringFlagPutHandler(logs.GlogSetter)))
 	}
 
 	// Start.

pkg/utils/osx/file.go

@@ -30,6 +30,14 @@ func Open(path string) (*os.File, error) {
 	return os.Open(p)
 }
 
+// ExistsParentDir checks if the parent directory of the given path exists.
+func ExistsParentDir(path string) bool {
+	p := filepath.Clean(path)
+	p = InlineTilde(p)
+
+	return ExistsDir(filepath.Dir(p))
+}
+
 // Exists checks if the given path exists.
 func Exists(path string, checks ...func(os.FileInfo) bool) bool {
 	p := filepath.Clean(path)
@@ -222,14 +230,94 @@ func IsEmptyFile(file string) bool {
 	return s.Size() == 0
 }
 
+// Remove removes the given file or directory,
+// and also supports additional checks before removal.
+func Remove(path string, checks ...func(os.FileInfo) error) error {
+	p := filepath.Clean(path)
+	p = InlineTilde(p)
+
+	stat, err := os.Lstat(p)
+	if err != nil {
+		if os.IsNotExist(err) {
+			return nil
+		}
+		return err
+	}
+
+	for i := range checks {
+		if checks[i] == nil {
+			continue
+		}
+
+		if err = checks[i](stat); err != nil {
+			return err
+		}
+	}
+
+	return os.Remove(p)
+}
+
+// RemoveDir removes the given directory, and returns an error if the path is not a directory.
+func RemoveDir(path string) error {
+	return Remove(path, func(stat os.FileInfo) error {
+		if !stat.Mode().IsDir() {
+			return fmt.Errorf("not a directory")
+		}
+		return nil
+	})
+}
+
+// RemoveFile removes the given file, and returns an error if the path is not a regular file.
+func RemoveFile(path string) error {
+	return Remove(path, func(stat os.FileInfo) error {
+		if !stat.Mode().IsRegular() {
+			return fmt.Errorf("not a regular file")
+		}
+		return nil
+	})
+}
+
+// RemoveLink removes the given symbolic link, and returns an error if the path is not a symbolic link.
+func RemoveLink(path string) error {
+	return Remove(path, func(stat os.FileInfo) error {
+		if stat.Mode()&os.ModeSymlink == 0 {
+			return fmt.Errorf("not a symbolic link")
+		}
+		return nil
+	})
+}
+
+// RemoveSocket removes the given socket, and returns an error if the path is not a socket.
+func RemoveSocket(path string) error {
+	return Remove(path, func(stat os.FileInfo) error {
+		if stat.Mode()&os.ModeSocket == 0 {
+			return fmt.Errorf("not a socket")
+		}
+		return nil
+	})
+}
+
+// RemoveDevice removes the given device, and returns an error if the path is not a device.
+func RemoveDevice(path string) error {
+	return Remove(path, func(stat os.FileInfo) error {
+		if stat.Mode()&os.ModeDevice == 0 {
+			return fmt.Errorf("not a device")
+		}
+		return nil
+	})
+}
+
 // DurableRemove removes the given file or directory,
 // and also syncs the parent directory to ensure the removal is durable.
 func DurableRemove(path string) error {
-	err := os.Remove(path)
+	p := filepath.Clean(path)
+	p = InlineTilde(p)
+
+	err := os.Remove(p)
 	if err != nil {
 		return err
 	}
-	return syncDir(filepath.Dir(path))
+	return syncDir(filepath.Dir(p))
 }
 
 func syncDir(dir string) error {

pkg/webserver/null.go

@@ -40,3 +40,6 @@ func (null) WebhookMux() *http.ServeMux {
 func (null) HostPort() (string, int, error) {
 	return "", 0, errors.New("no listener")
 }
+
+func (null) NotFoundHandler(http.Handler) {
+}

pkg/webserver/option.go

@@ -18,10 +18,13 @@ import (
 )
 
 type Options struct {
+	FlagOptions
+
 	// Establish.
-	BindAddress net.IP
-	BindPort    int
-	CertDir     string
+	BindUnixPath string
+	BindAddress  net.IP
+	BindPort     int
+	CertDir      string
 }
 
 func NewOptions() *Options {
@@ -32,8 +35,31 @@ func NewOptions() *Options {
 	}
 }
 
-func (o *Options) AddFlags(fs *pflag.FlagSet) {
+type (
+	FlagOptions struct {
+		noBindUnixPath bool
+	}
+
+	FlagOption func(opts *FlagOptions)
+)
+
+func WithoutBindUnixPath() FlagOption {
+	return func(opts *FlagOptions) {
+		opts.noBindUnixPath = true
+	}
+}
+
+func (o *Options) AddFlags(fs *pflag.FlagSet, opts ...FlagOption) {
+	for i := range opts {
+		opts[i](&o.FlagOptions)
+	}
+
 	// Establish.
+	if !o.noBindUnixPath {
+		fs.StringVar(&o.BindUnixPath, "bind-unix-path", o.BindUnixPath,
+			"the unix socket path on which to serve. "+
+				"if specified, --bind-address and --secure-port will be ignored.")
+	}
 	fs.IPVar(&o.BindAddress, "bind-address", o.BindAddress,
 		"the IP address(without port) on which to serve.")
 	fs.IntVar(&o.BindPort, "secure-port", o.BindPort,
@@ -45,6 +71,12 @@ func (o *Options) AddFlags(fs *pflag.FlagSet) {
 
 func (o *Options) Validate(_ context.Context) error {
 	// Establish.
+	if !o.noBindUnixPath && o.BindUnixPath != "" {
+		if !osx.ExistsParentDir(o.BindUnixPath) {
+			return errors.New("--bind-unix-path: no found parent directory")
+		}
+		return nil
+	}
 	if o.BindPort < 1 || o.BindPort > 65535 {
 		return errors.New("--secure-port: out of range")
 	}
@@ -91,11 +123,24 @@ func (o *Options) Complete(_ context.Context) (*Config, error) {
 		tlsCfg.GetCertificate = certMgr.GetCertificate
 	}
 
-	listener, err := tls.Listen("tcp", net.JoinHostPort(o.BindAddress.String(), strconv.Itoa(o.BindPort)), tlsCfg)
+	var (
+		lis net.Listener
+		err error
+	)
+	if o.BindUnixPath != "" {
+		err = osx.RemoveSocket(o.BindUnixPath)
+		if err != nil {
+			return nil, fmt.Errorf("remove existing unix socket: %w", err)
+		}
+		lis, err = tls.Listen("unix", o.BindUnixPath, tlsCfg)
+	} else {
+		lis, err = tls.Listen("tcp",
+			net.JoinHostPort(o.BindAddress.String(), strconv.Itoa(o.BindPort)), tlsCfg)
+	}
 	if err != nil {
 		return nil, fmt.Errorf("create listener: %w", err)
 	}
 
-	cfg.Listener = listener
+	cfg.Listener = lis
 	return &cfg, nil
 }

pkg/webserver/server.go

@@ -16,10 +16,16 @@ import (
 	"gpustack.ai/gpustack/pkg/utils/netx"
 )
 
+// Server is a web server that can serve webhooks and other HTTP handlers.
 type Server interface {
 	ctrlwebhook.Server
 
+	// HostPort returns the host and port that the server is listening on.
 	HostPort() (string, int, error)
+
+	// NotFoundHandler sets the handler for requests that do not match any registered path.
+	// If not set, the default http.NotFoundHandler will be used.
+	NotFoundHandler(http.Handler)
 }
 
 // New creates a Server with the given configuration.
@@ -32,7 +38,7 @@ func New(c *Config) (Server, error) {
 		return nil, errors.New("listener must be a TCP listener")
 	}
 
-	return server{
+	return &server{
 		listener: c.Listener,
 		host:     tcpAddr.IP.String(),
 		port:     tcpAddr.Port,
@@ -42,23 +48,24 @@ func New(c *Config) (Server, error) {
 }
 
 type server struct {
-	listener net.Listener
-	host     string
-	port     int
-	runners  []manager.Runnable
-	mux      *http.ServeMux
+	listener        net.Listener
+	host            string
+	port            int
+	runners         []manager.Runnable
+	mux             *http.ServeMux
+	notFoundHandler http.Handler
 }
 
-func (server) NeedLeaderElection() bool {
+func (*server) NeedLeaderElection() bool {
 	return false
 }
 
-func (s server) Register(path string, hook http.Handler) {
+func (s *server) Register(path string, hook http.Handler) {
 	s.mux.Handle(path, hook)
 }
 
-func (s server) Start(ctx context.Context) error {
-	srv := newHttpServer(s.mux)
+func (s *server) Start(ctx context.Context) error {
+	srv := s.getServer()
 
 	gp := gox.GroupWithContextIn(ctx)
 	for i := range s.runners {
@@ -84,17 +91,43 @@ func (s server) Start(ctx context.Context) error {
 	return gp.Wait()
 }
 
-func (s server) StartedChecker() healthz.Checker {
+func (s *server) StartedChecker() healthz.Checker {
 	addr := net.JoinHostPort(s.host, strconv.Itoa(s.port))
 	return func(req *http.Request) error {
 		return netx.IsConnected(req.Context(), "tls", addr, 10*time.Second)
 	}
 }
 
-func (s server) WebhookMux() *http.ServeMux {
+func (s *server) WebhookMux() *http.ServeMux {
 	return s.mux
 }
 
-func (s server) HostPort() (string, int, error) {
+func (s *server) HostPort() (string, int, error) {
 	return s.host, s.port, nil
 }
+
+func (s *server) NotFoundHandler(notFoundHandler http.Handler) {
+	s.notFoundHandler = notFoundHandler
+}
+
+func (s *server) getNotFoundHandler() http.Handler {
+	if s.notFoundHandler == nil {
+		return http.NotFoundHandler()
+	}
+	return s.notFoundHandler
+}
+
+func (s *server) getServer() *http.Server {
+	nf := s.getNotFoundHandler()
+
+	h := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		h, pattern := s.mux.Handler(r)
+		if pattern == "" {
+			nf.ServeHTTP(w, r)
+			return
+		}
+		h.ServeHTTP(w, r)
+	})
+
+	return newHttpServer(h)
+}

pkg/worker/cmd.go

@@ -13,7 +13,10 @@ func NewCommand() *cobra.Command {
 	o := NewOptions()
 
 	c := &cobra.Command{
-		Use:   "worker",
+		Use: "worker",
+		Aliases: []string{
+			"w",
+		},
 		Short: "assists in managing GPUStack Kubernetes resources.",
 		PreRunE: func(c *cobra.Command, args []string) error {
 			return o.Validate(c.Context())