refactor(dynamic mount): introduce dynamic server manager

To ensure secure isolation for each dynamic mount and avoid unstable
mount propagation, an independent csi.sock is currently created under
each dynamic mount directory instead of using a shared csi.sock, these
individual csi.sock servers are managed by the DynamicServerManager.

Signed-off-by: imeoer <[email protected]>

Author: imeoer

pkg/config/config.go

@@ -35,19 +35,23 @@ type RawConfig struct {
 	// 	static: /var/lib/dragonfly/model-csi/volumes/$volumeName/model
 	// dynamic: /var/lib/dragonfly/model-csi/volumes/$volumeName/models
 	//          /var/lib/dragonfly/model-csi/volumes/$volumeName/csi.sock
-	ServiceName              string     `yaml:"service_name"`
-	RootDir                  string     `yaml:"root_dir"`
-	ExternalCSIEndpoint      string     `yaml:"external_csi_endpoint"`
-	ExternalCSIAuthorization string     `yaml:"external_csi_authorization"`
-	DynamicCSIEndpoint       string     `yaml:"dynamic_csi_endpoint"`
-	CSIEndpoint              string     `yaml:"csi_endpoint"`
-	MetricsAddr              string     `yaml:"metrics_addr"`
-	TraceEndpoint            string     `yaml:"trace_endpoint"`
-	PprofAddr                string     `yaml:"pprof_addr"`
-	PullConfig               PullConfig `yaml:"pull_config"`
-	Features                 Features   `yaml:"features"`
-	NodeID                   string     // From env CSI_NODE_ID
-	Mode                     string     // From env X_CSI_MODE: "controller" or "node"
+	ServiceName              string `yaml:"service_name"`
+	RootDir                  string `yaml:"root_dir"`
+	ExternalCSIEndpoint      string `yaml:"external_csi_endpoint"`
+	ExternalCSIAuthorization string `yaml:"external_csi_authorization"`
+	// Deprecated: To ensure secure isolation for each dynamic mount and avoid
+	// unstable mount propagation, an independent csi.sock is currently created
+	// under each dynamic mount directory instead of using a shared csi.sock,
+	// these individual csi.sock servers are managed by the DynamicServerManager.
+	DynamicCSIEndpoint string     `yaml:"dynamic_csi_endpoint"`
+	CSIEndpoint        string     `yaml:"csi_endpoint"`
+	MetricsAddr        string     `yaml:"metrics_addr"`
+	TraceEndpoint      string     `yaml:"trace_endpoint"`
+	PprofAddr          string     `yaml:"pprof_addr"`
+	PullConfig         PullConfig `yaml:"pull_config"`
+	Features           Features   `yaml:"features"`
+	NodeID             string     // From env CSI_NODE_ID
+	Mode               string     // From env X_CSI_MODE: "controller" or "node"
 }
 
 type Features struct {
@@ -135,6 +139,11 @@ func (cfg *RawConfig) GetCSISockDirForDynamic(volumeName string) string {
 	return filepath.Join(cfg.GetVolumeDirForDynamic(volumeName), "csi")
 }
 
+// /var/lib/dragonfly/model-csi/volumes/$volumeName/csi/csi.sock
+func (cfg *RawConfig) GetCSISockPathForDynamic(volumeName string) string {
+	return filepath.Join(cfg.GetCSISockDirForDynamic(volumeName), "csi.sock")
+}
+
 func (cfg *RawConfig) IsControllerMode() bool {
 	return cfg.Mode == "controller"
 }

pkg/server/http.go

@@ -6,7 +6,6 @@ import (
 	"net/http"
 	"net/url"
 	"os"
-	"path/filepath"
 	"strings"
 	"time"
 
@@ -27,6 +26,7 @@ import (
 	"github.com/modelpack/model-csi-driver/pkg/metrics"
 	"github.com/modelpack/model-csi-driver/pkg/provider"
 	"github.com/modelpack/model-csi-driver/pkg/service"
+	"github.com/modelpack/model-csi-driver/pkg/utils"
 	"go.opentelemetry.io/contrib/instrumentation/google.golang.org/grpc/otelgrpc"
 )
 
@@ -42,20 +42,6 @@ var kasp = keepalive.ServerParameters{
 	Timeout: 30 * time.Second, // Wait 30 second for the ping ack before assuming the connection is dead
 }
 
-func ensureSockNotExists(ctx context.Context, sockPath string) error {
-	_, err := os.Stat(sockPath)
-	if err == nil {
-		if err = os.Remove(sockPath); err != nil {
-			return errors.Wrapf(err, "remove existed sock path: %s", sockPath)
-		}
-		logger.WithContext(ctx).Infof("removed existed sock path: %s", sockPath)
-	}
-	if err = os.MkdirAll(filepath.Dir(sockPath), 0755); err != nil {
-		return errors.Wrapf(err, "create sock path dir: %s", filepath.Dir(sockPath))
-	}
-	return nil
-}
-
 func isSockListening(sockPath string) bool {
 	if _, err := os.Stat(sockPath); err != nil {
 		return false
@@ -157,7 +143,7 @@ func (server *Server) Run(ctx context.Context) error {
 			return errors.Wrap(err, "parse external csi endpoint")
 		}
 		if endpoint.Path != "" {
-			if err := ensureSockNotExists(ctx, endpoint.Path); err != nil {
+			if err := utils.EnsureSockNotExists(ctx, endpoint.Path); err != nil {
 				return errors.Wrapf(err, "ensure socket not exists: %s", endpoint.Path)
 			}
 		}
@@ -243,26 +229,28 @@ func (server *Server) Run(ctx context.Context) error {
 			}))
 		}
 
+		// nolint:staticcheck
 		if server.cfg.Get().DynamicCSIEndpoint != "" {
 			eg.Go(withFatalError(func() error {
+				if err := server.svc.DynamicServerManager.RecoverServers(context.Background()); err != nil {
+					return errors.Wrap(err, "recover dynamic http servers")
+				}
+
+				// Deprecated: use DynamicServerManager to manage dynamic csi.sock servers,
+				// keep this for backward compatibility.
+				// nolint:staticcheck
 				endpoint, err := url.Parse(server.cfg.Get().DynamicCSIEndpoint)
 				if err != nil {
 					return errors.Wrap(err, "parse dynamic csi endpoint")
 				}
 				if endpoint.Path != "" {
-					if err := ensureSockNotExists(ctx, endpoint.Path); err != nil {
-						return errors.Wrapf(err, "ensure socket not exists: %s", endpoint.Path)
+					_, err = server.svc.DynamicServerManager.CreateServer(context.Background(), endpoint.Path)
+					if err != nil {
+						return errors.Wrap(err, "create dynamic http server")
 					}
 				}
 
-				logger.WithContext(ctx).Infof("serving dynamic http server on %s", server.cfg.Get().DynamicCSIEndpoint)
-
-				httpServer, err := NewHTTPServer(server.cfg, server.svc)
-				if err != nil {
-					return errors.Wrap(err, "create dynamic http server")
-				}
-
-				return httpServer.Serve()
+				return nil
 			}))
 		}
 	}

pkg/server/server.go

@@ -3,30 +3,22 @@ package server
 import (
 	"context"
 	"fmt"
-	"net/http"
-	"net/http/httptest"
-	"net/url"
 	"os"
 	"os/exec"
 	"path/filepath"
 	"strings"
-	"syscall"
 	"testing"
 	"time"
 
-	"github.com/labstack/echo/v4"
 	"github.com/modelpack/model-csi-driver/pkg/client"
 	"github.com/modelpack/model-csi-driver/pkg/config"
 	"github.com/modelpack/model-csi-driver/pkg/service"
 	"github.com/modelpack/model-csi-driver/pkg/status"
 	modelspec "github.com/modelpack/model-spec/specs-go/v1"
 	"github.com/opencontainers/go-digest"
 	ocispec "github.com/opencontainers/image-spec/specs-go/v1"
-	"github.com/pkg/errors"
 	"github.com/stretchr/testify/require"
 	"golang.org/x/sync/errgroup"
-	"google.golang.org/grpc/codes"
-	grpcStatus "google.golang.org/grpc/status"
 )
 
 const (
@@ -593,16 +585,3 @@ func TestServer(t *testing.T) {
 
 	run(t, "curl http://127.0.0.1:5244/metrics | grep -v '# '")
 }
-
-func TestHandleError(t *testing.T) {
-	echo := echo.New()
-	recorder := httptest.NewRecorder()
-	echoCtx := echo.NewContext(&http.Request{URL: &url.URL{RawQuery: ""}}, recorder)
-	err := handleError(echoCtx, grpcStatus.Error(codes.ResourceExhausted, errors.Wrap(errors.Wrapf(errors.Wrapf(syscall.ENOSPC, "model image  is , but only of disk quota is available"), "pull model failed"), "pull model for dynamic volume").Error()))
-	require.NoError(t, err)
-	require.Equal(t, http.StatusNotAcceptable, echoCtx.Response().Status)
-
-	err = handleError(echoCtx, grpcStatus.Error(codes.FailedPrecondition, "error test"))
-	require.NoError(t, err)
-	require.Equal(t, http.StatusInternalServerError, echoCtx.Response().Status)
-}