Merge pull request #1 from bachue/fix/data-corruption

bachue · web-flow · commit 7f32413df68b · 2022-11-23T10:48:35.000+08:00
Fix possible data corruption
diff --git a/connector/main.go b/connector/main.go
@@ -80,23 +80,32 @@ func main() {
 		os.Exit(1)
 	}
 
-	homeDir, err := os.UserHomeDir()
-	if err != nil {
-		homeDir = os.TempDir()
+	if userConfigDir, err := os.UserConfigDir(); err != nil {
+		rcloneConfigDir = filepath.Join(os.TempDir(), ".rclone", "config")
+	} else {
+		rcloneConfigDir = filepath.Join(userConfigDir, "rclone")
+	}
+
+	if userCacheDir, err := os.UserCacheDir(); err != nil {
+		rcloneCacheDir = filepath.Join(os.TempDir(), ".rclone", "cache")
+	} else {
+		rcloneCacheDir = filepath.Join(userCacheDir, "rclone")
+	}
+
+	if userLogDir, err := userLogDir(); err != nil {
+		rcloneLogDir = filepath.Join(os.TempDir(), ".rclone", "log")
+	} else {
+		rcloneLogDir = filepath.Join(userLogDir, "rclone")
 	}
-	rcloneConfigDir = filepath.Join(homeDir, ".rclone", "config")
+
 	if err = ensureDirectoryExists(rcloneConfigDir); err != nil {
 		fmt.Fprintf(os.Stderr, "Failed to ensure directory %s exists: %s", rcloneConfigDir, err)
 		os.Exit(1)
 	}
-
-	rcloneCacheDir = filepath.Join(homeDir, ".rclone", "cache")
 	if err = ensureDirectoryExists(rcloneCacheDir); err != nil {
 		fmt.Fprintf(os.Stderr, "Failed to ensure directory %s exists: %s", rcloneCacheDir, err)
 		os.Exit(1)
 	}
-
-	rcloneLogDir = filepath.Join(homeDir, ".rclone", "log")
 	if err = ensureDirectoryExists(rcloneLogDir); err != nil {
 		fmt.Fprintf(os.Stderr, "Failed to ensure directory %s exists: %s", rcloneLogDir, err)
 		os.Exit(1)
@@ -270,6 +279,15 @@ func handleConn(conn net.Conn, cmdIn <-chan protocol.Cmd, cmdOut chan<- protocol
 				log.Infof("Received requestDataCmd: %#v", payload)
 				cmdOut <- payload
 			}
+		case protocol.KodoUmountCmdName:
+			payload := new(protocol.KodoUmountCmd)
+			if err := json.Unmarshal([]byte(request.Payload), payload); err != nil {
+				log.Warnf("Protocol %s payload parse error: %s", request.Cmd, err)
+				return
+			} else {
+				log.Infof("Received kodoUmountCmd: %#v", payload)
+				cmdOut <- payload
+			}
 		default:
 			log.Warnf("Unrecognized request cmd: %s", request.Cmd)
 			return
@@ -351,7 +369,7 @@ func handleCmd(cmdOut chan<- protocol.Cmd, cmdIn <-chan protocol.Cmd) {
 			log.Errorf("Failed to create stderr pipe: %s", err)
 			return false
 		}
-		go outputReader("stderr", stderr, false)
+		go outputReader("stderr", stderr, true)
 		go func() {
 			defer cancel()
 			err := execCmd.Run()
@@ -389,13 +407,32 @@ func handleCmd(cmdOut chan<- protocol.Cmd, cmdIn <-chan protocol.Cmd) {
 					log.Warnf("Failed to write rclone config: %s", err)
 					return
 				}
+				uuid := rcloneCacheId(c.MountPath)
+				volumeCacheDir := filepath.Join(rcloneCacheDir, c.VolumeId, uuid)
+				if err = ensureDirectoryExists(volumeCacheDir); err != nil {
+					log.Errorf("Failed to ensure directory %s exists: %s", volumeCacheDir, err)
+					return
+				}
+				rcloneLogFile := filepath.Join(rcloneLogDir, c.VolumeId, uuid+".log")
+				if err = ensureDirectoryExists(filepath.Dir(rcloneLogFile)); err != nil {
+					log.Errorf("Failed to ensure directory %s exists: %s", filepath.Dir(rcloneLogFile), err)
+					return
+				}
 				ctx = context.WithValue(ctx, protocol.ContextKeyConfigFilePath, rcloneConfigPath)
 				ctx = context.WithValue(ctx, protocol.ContextKeyUserAgent, userAgent)
-				ctx = context.WithValue(ctx, protocol.ContextKeyLogDirPath, rcloneLogDir)
-				ctx = context.WithValue(ctx, protocol.ContextKeyCacheDirPath, rcloneCacheDir)
+				ctx = context.WithValue(ctx, protocol.ContextKeyLogFilePath, rcloneLogFile)
+				ctx = context.WithValue(ctx, protocol.ContextKeyCacheDirPath, volumeCacheDir)
 				if ok := execCommand(c.ExecCommand(ctx), func() { os.Remove(rcloneConfigPath) }); !ok {
 					return
 				}
+			case *protocol.KodoUmountCmd:
+				uuid := rcloneCacheId(c.MountPath)
+				volumeCacheDir := filepath.Join(rcloneCacheDir, c.VolumeId, uuid)
+				rcloneLogFile := filepath.Join(rcloneLogDir, c.VolumeId, uuid+".log")
+				os.RemoveAll(volumeCacheDir)
+				os.Remove(rcloneLogFile)
+				os.Remove(filepath.Dir(rcloneLogFile))
+				os.Remove(filepath.Dir(volumeCacheDir))
 			case *protocol.RequestDataCmd:
 				if stdin == nil {
 					log.Warnf("Received RequestDataCmd when process is not started")
diff --git a/connector/utils.go b/connector/utils.go
@@ -2,11 +2,16 @@ package main
 
 import (
 	"bytes"
+	"crypto/md5"
+	"encoding/hex"
+	"errors"
 	"fmt"
+	"io"
 	"os"
 	"os/exec"
 	"path/filepath"
 	"regexp"
+	"runtime"
 	"strconv"
 
 	"github.com/Unknwon/goconfig"
@@ -34,6 +39,46 @@ const (
 	RCLONE_CONFIG_BOOL_TRUE             = "true"
 )
 
+func userLogDir() (string, error) {
+	var dir string
+
+	switch runtime.GOOS {
+	case "windows":
+		dir = os.Getenv("LocalAppData")
+		if dir == "" {
+			return "", errors.New("%LocalAppData% is not defined")
+		}
+
+	case "darwin", "ios":
+		dir = os.Getenv("HOME")
+		if dir == "" {
+			return "", errors.New("$HOME is not defined")
+		}
+		dir += "/Library/Logs"
+
+	case "plan9":
+		dir = os.Getenv("home")
+		if dir == "" {
+			return "", errors.New("$home is not defined")
+		}
+		dir += "/lib/log"
+
+	default: // Unix
+		dir = "/var/log"
+	}
+
+	return dir, nil
+}
+
+func rcloneCacheId(items ...string) string {
+	hasher := md5.New()
+	for _, item := range items {
+		io.WriteString(hasher, item)
+		hasher.Write([]byte{0})
+	}
+	return hex.EncodeToString(hasher.Sum(nil))
+}
+
 func ensureDirectoryExists(path string) error {
 	if fileInfo, err := os.Stat(path); os.IsNotExist(err) {
 		return os.MkdirAll(path, 0700)
diff --git a/docker/Dockerfile b/docker/Dockerfile
@@ -1,7 +1,7 @@
 FROM debian:bullseye
 
 ARG KODOFS_VERSION=2.4.18
-ARG RCLONE_VERSION=1.60.0
+ARG RCLONE_VERSION=1.60.1
 COPY nsenter /usr/local/bin/nsenter
 COPY kodofs-v${KODOFS_VERSION} /usr/local/bin/kodofs
 COPY rclone-v${RCLONE_VERSION} /usr/local/bin/rclone
diff --git a/docker/rclone-v1.60.1 b/docker/rclone-v1.60.1
diff --git a/k8s/kodo/kodo-plugin.yaml b/k8s/kodo/kodo-plugin.yaml
@@ -57,7 +57,7 @@ spec:
             capabilities:
               add: ["SYS_ADMIN"]
             allowPrivilegeEscalation: true
-          image: kodoproduct/csi-plugin.storage.qiniu.com:v0.1.0
+          image: kodoproduct/csi-plugin.storage.qiniu.com:v0.1.0-5-g9007c38
           imagePullPolicy: Always
           args:
             - "--endpoint=$(CSI_ENDPOINT)"
diff --git a/k8s/kodofs/kodofs-plugin.yaml b/k8s/kodofs/kodofs-plugin.yaml
@@ -57,7 +57,7 @@ spec:
             capabilities:
               add: ["SYS_ADMIN"]
             allowPrivilegeEscalation: true
-          image: kodoproduct/csi-plugin.storage.qiniu.com:v0.1.0
+          image: kodoproduct/csi-plugin.storage.qiniu.com:v0.1.0-5-g9007c38
           imagePullPolicy: Always
           args:
             - "--endpoint=$(CSI_ENDPOINT)"
diff --git a/plugin/kodo_node_server.go b/plugin/kodo_node_server.go
@@ -70,6 +70,11 @@ func (server *kodoNodeServer) NodeUnpublishVolume(ctx context.Context, req *csi.
 	} else {
 		log.Infof("NodeUnpublishVolume: umounted kodo volume from path: %s", mountPath)
 	}
+	if err = cleanAfterKodoUmount(req.VolumeId, mountPath); err != nil {
+		log.Warnf("NodeUnpublishVolume: failed to clean kodo volume cache and log files: %s", err)
+	} else {
+		log.Infof("NodeUnpublishVolume: kodo volume cache and log files are cleaned")
+	}
 	return &csi.NodeUnpublishVolumeResponse{}, nil
 }
 
diff --git a/plugin/utils.go b/plugin/utils.go
@@ -189,19 +189,11 @@ func mountKodoFS(gatewayID, mountPath string, mountServerAddress *url.URL, acces
 		}
 		switch cmd.(type) {
 		case *protocol.InitKodoFSMountCmd:
-			if err = encoder.Encode(&protocol.Request{
-				Version: protocol.Version,
-				Cmd:     protocol.InitKodoFsMountCmdName,
-				Payload: json.RawMessage(buf),
-			}); err != nil {
+			if err = encoder.Encode(makeRequest(protocol.InitKodoFsMountCmdName, buf)); err != nil {
 				return fmt.Errorf("failed to write command to unix socket %s: %w", SocketPath, err)
 			}
 		case *protocol.RequestDataCmd:
-			if err = encoder.Encode(&protocol.Request{
-				Version: protocol.Version,
-				Cmd:     protocol.RequestDataCmdName,
-				Payload: json.RawMessage(buf),
-			}); err != nil {
+			if err = encoder.Encode(makeRequest(protocol.RequestDataCmdName, buf)); err != nil {
 				return fmt.Errorf("failed to write command to unix socket %s: %w", SocketPath, err)
 			}
 		}
@@ -287,11 +279,7 @@ func mountKodo(volumeId, mountPath, subDir, accessKey, secretKey, bucketId, s3Re
 		}
 		switch cmd.(type) {
 		case *protocol.InitKodoMountCmd:
-			if err = encoder.Encode(&protocol.Request{
-				Version: protocol.Version,
-				Cmd:     protocol.InitKodoMountCmdName,
-				Payload: json.RawMessage(buf),
-			}); err != nil {
+			if err = encoder.Encode(makeRequest(protocol.InitKodoMountCmdName, buf)); err != nil {
 				return fmt.Errorf("failed to write command to unix socket %s: %w", SocketPath, err)
 			}
 		}
@@ -411,6 +399,44 @@ func umount(mountPath string) error {
 	return err
 }
 
+func cleanAfterKodoUmount(volumeId, mountPath string) error {
+	conn, err := net.Dial("unix", SocketPath)
+	if err != nil {
+		return fmt.Errorf("failed to dial unix socket %s: %w", SocketPath, err)
+	}
+	defer conn.Close()
+
+	encoder := json.NewEncoder(conn)
+
+	writeCmdToConn := func(encoder *json.Encoder, cmd protocol.Cmd) error {
+		buf, err := json.Marshal(cmd)
+		if err != nil {
+			return fmt.Errorf("failed to marshal json payload: %w", err)
+		}
+		switch cmd.(type) {
+		case *protocol.KodoUmountCmd:
+			if err = encoder.Encode(makeRequest(protocol.KodoUmountCmdName, buf)); err != nil {
+				return fmt.Errorf("failed to write command to unix socket %s: %w", SocketPath, err)
+			}
+		}
+		return nil
+	}
+
+	cmd := protocol.KodoUmountCmd{
+		VolumeId:  volumeId,
+		MountPath: mountPath,
+	}
+	return writeCmdToConn(encoder, &cmd)
+}
+
+func makeRequest(cmdName string, buf []byte) *protocol.Request {
+	return &protocol.Request{
+		Version: protocol.Version,
+		Cmd:     cmdName,
+		Payload: json.RawMessage(buf),
+	}
+}
+
 const (
 	FuseTypeKodoFS = "fuse.KodoFS"
 	FuseTypeKodo   = "fuse.rclone"
diff --git a/protocol/protocol.go b/protocol/protocol.go
@@ -5,14 +5,14 @@ import (
 	"encoding/json"
 	"fmt"
 	"os/exec"
-	"path/filepath"
 	"strconv"
 )
 
 const (
 	Version                = "v2"
 	InitKodoMountCmdName   = "init_kodo_mount"
 	InitKodoFsMountCmdName = "init_kodofs_mount"
+	KodoUmountCmdName      = "umount_kodo"
 	RequestDataCmdName     = "request_data"
 	ResponseDataCmdName    = "response_data"
 	TerminateCmdName       = "terminate"
@@ -68,6 +68,11 @@ type (
 		DebugFuse             bool    `json:"debug_fuse,omitempty"`
 	}
 
+	KodoUmountCmd struct {
+		VolumeId  string `json:"volume_id"`
+		MountPath string `json:"mount_path"`
+	}
+
 	RequestDataCmd struct {
 		Data string `json:"data"`
 	}
@@ -93,6 +98,7 @@ type (
 
 func (*InitKodoFSMountCmd) Command() {}
 func (*InitKodoMountCmd) Command()   {}
+func (*KodoUmountCmd) Command()      {}
 func (*RequestDataCmd) Command()     {}
 func (*ResponseDataCmd) Command()    {}
 func (*TerminateCmd) Command()       {}
@@ -107,7 +113,7 @@ const (
 
 	ContextKeyConfigFilePath contextKey = "config_file_path"
 	ContextKeyUserAgent      contextKey = "user_agent"
-	ContextKeyLogDirPath     contextKey = "log_dir_path"
+	ContextKeyLogFilePath    contextKey = "log_file_path"
 	ContextKeyCacheDirPath   contextKey = "cache_dir_path"
 )
 
@@ -119,14 +125,14 @@ func (c *InitKodoFSMountCmd) ExecCommand(ctx context.Context) *exec.Cmd {
 func (c *InitKodoMountCmd) ExecCommand(ctx context.Context) *exec.Cmd {
 	rcloneConfigFilePath := ctx.Value(ContextKeyConfigFilePath).(string)
 	userAgent := ctx.Value(ContextKeyUserAgent).(string)
-	rcloneLogDirPath := ctx.Value(ContextKeyLogDirPath).(string)
+	rcloneLogFilePath := ctx.Value(ContextKeyLogFilePath).(string)
 	rcloneCacheDirPath := ctx.Value(ContextKeyCacheDirPath).(string)
 
 	var cmdFlags = []string{
 		"--auto-confirm",
 		"--config", rcloneConfigFilePath,
 		"--user-agent", fmt.Sprintf("%s/%s", userAgent, c.VolumeId),
-		"--log-file", filepath.Join(rcloneLogDirPath, c.VolumeId+".log")}
+		"--log-file", rcloneLogFilePath}
 	if c.BufferSize != nil {
 		cmdFlags = append(cmdFlags, []string{"--buffer-size", formatByteSize(*c.BufferSize)}...)
 	}

Original file line number	Diff line number	Diff line change
`@@ -70,6 +70,11 @@ func (server kodoNodeServer) NodeUnpublishVolume(ctx context.Context, req csi.`
`70`	`70`	`} else {`
`71`	`71`	`log.Infof("NodeUnpublishVolume: umounted kodo volume from path: %s", mountPath)`
`72`	`72`	`}`
	`73`	`+ if err = cleanAfterKodoUmount(req.VolumeId, mountPath); err != nil {`
	`74`	`+ log.Warnf("NodeUnpublishVolume: failed to clean kodo volume cache and log files: %s", err)`
	`75`	`+ } else {`
	`76`	`+ log.Infof("NodeUnpublishVolume: kodo volume cache and log files are cleaned")`
	`77`	`+ }`
`73`	`78`	`return &csi.NodeUnpublishVolumeResponse{}, nil`
`74`	`79`	`}`
`75`	`80`