CON-9040 driver can accidentally format existing volume

README.md

@@ -309,6 +309,16 @@ For that reason, the default page size can be customized by passing the `--defau
 
 DO API usage is subject to [certain rate limits](https://docs.digitalocean.com/reference/api/api-reference/#section/Introduction/Rate-Limit). In order to protect against running out of quota for extremely heavy regular usage or pathological cases (e.g., bugs or API thrashing due to an interfering third-party controller), a custom rate limit can be configured via the `--do-api-rate-limit` flag. It accepts a float value, e.g., `--do-api-rate-limit=3.5` to restrict API usage to 3.5 queries per second.
 
+### Flags
+
+| Name                  | Description                                                                          | Default |
+|-----------------------|--------------------------------------------------------------------------------------|---------|
+| --validate-attachment | Validate if the attachment has fully completed before formatting/mounting the device | false   |
+
+The `--validate-attachment` options adds an additional validation which checks for the `/sys/class/block/<device name>/device/state`
+file content for the `running` status. When enabling this flag, it prevents a racing condition where the DOBS volumes aren't 
+fully attached which can be misinterpreted by the CSI implementation causing a force format of the volume which results in data loss. 
+
 ---
 
 ## Development

cmd/do-csi-plugin/Dockerfile

@@ -12,7 +12,7 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
-FROM alpine:3.16
+FROM amd64/alpine:3.16
 
 # e2fsprogs-extra is required for resize2fs used for the resize operation
 # blkid: block device identification tool from util-linux

cmd/do-csi-plugin/main.go

@@ -39,6 +39,7 @@ func main() {
 		debugAddr              = flag.String("debug-addr", "", "Address to serve the HTTP debug server on.")
 		defaultVolumesPageSize = flag.Uint("default-volumes-page-size", 0, "The default page size used when paging through volumes results (default: do not specify and let the DO API choose)")
 		doAPIRateLimitQPS      = flag.Float64("do-api-rate-limit", 0, "Impose QPS rate limit on DigitalOcean API usage (default: do not rate limit)")
+		validateAttachment     = flag.Bool("validate-attachment", false, "Validate if the attachment has fully completed before formatting/mounting the device")
 		version                = flag.Bool("version", false, "Print the version and exit.")
 	)
 	flag.Parse()
@@ -62,6 +63,7 @@ func main() {
 		DebugAddr:              *debugAddr,
 		DefaultVolumesPageSize: *defaultVolumesPageSize,
 		DOAPIRateLimitQPS:      *doAPIRateLimitQPS,
+		ValidateAttachment:     *validateAttachment,
 	})
 	if err != nil {
 		log.Fatalln(err)

driver/driver.go

@@ -68,6 +68,7 @@ type Driver struct {
 	doTag                  string
 	isController           bool
 	defaultVolumesPageSize uint
+	validateAttachment     bool
 
 	srv     *grpc.Server
 	httpSrv *http.Server
@@ -100,6 +101,7 @@ type NewDriverParams struct {
 	DebugAddr              string
 	DefaultVolumesPageSize uint
 	DOAPIRateLimitQPS      float64
+	ValidateAttachment     bool
 }
 
 // NewDriver returns a CSI plugin that contains the necessary gRPC

driver/driver_test.go

@@ -539,6 +539,10 @@ func (f *fakeMounter) GetDeviceName(_ mount.Interface, mountPath string) (string
 	return "", nil
 }
 
+func (f *fakeMounter) IsAttached(source string) (bool, error) {
+	return true, nil
+}
+
 func (f *fakeMounter) IsFormatted(source string) (bool, error) {
 	return true, nil
 }

driver/mounter.go

@@ -33,6 +33,25 @@ import (
 	kexec "k8s.io/utils/exec"
 )
 
+const (
+	runningState = "running"
+)
+
+type prodAttachmentValidator struct{}
+
+func (av *prodAttachmentValidator) readFile(name string) ([]byte, error) {
+	return os.ReadFile(name)
+}
+
+func (av *prodAttachmentValidator) evalSymlinks(path string) (string, error) {
+	return filepath.EvalSymlinks(path)
+}
+
+type AttachmentValidator interface {
+	readFile(name string) ([]byte, error)
+	evalSymlinks(path string) (string, error)
+}
+
 type findmntResponse struct {
 	FileSystems []fileSystem `json:"filesystems"`
 }
@@ -67,6 +86,9 @@ type Mounter interface {
 	// Unmount unmounts the given target
 	Unmount(target string) error
 
+	// IsAttached checks whether the source device is in the running state.
+	IsAttached(source string) (bool, error)
+
 	// IsFormatted checks whether the source device is formatted or not. It
 	// returns true if the source device is already formatted.
 	IsFormatted(source string) (bool, error)
@@ -90,8 +112,9 @@ type Mounter interface {
 // architecture specific code in the future, such as mounter_darwin.go,
 // mounter_linux.go, etc..
 type mounter struct {
-	log      *logrus.Entry
-	kMounter *mount.SafeFormatAndMount
+	log                 *logrus.Entry
+	kMounter            *mount.SafeFormatAndMount
+	attachmentValidator AttachmentValidator
 }
 
 // newMounter returns a new mounter instance
@@ -102,8 +125,9 @@ func newMounter(log *logrus.Entry) *mounter {
 	}
 
 	return &mounter{
-		kMounter: kMounter,
-		log:      log,
+		kMounter:            kMounter,
+		log:                 log,
+		attachmentValidator: &(prodAttachmentValidator{}),
 	}
 }
 
@@ -208,6 +232,26 @@ func (m *mounter) Unmount(target string) error {
 	return mount.CleanupMountPoint(target, m.kMounter, true)
 }
 
+func (m *mounter) IsAttached(source string) (bool, error) {
+	out, err := m.attachmentValidator.evalSymlinks(source)
+	if err != nil {
+		return false, fmt.Errorf("error evaluating the symbolic link %q: %s", source, err)
+	}
+
+	_, deviceName := filepath.Split(out)
+	deviceStateFilePath := fmt.Sprintf("/sys/class/block/%s/device/state", deviceName)
+	deviceStateFileContent, err := m.attachmentValidator.readFile(deviceStateFilePath)
+	if err != nil {
+		return false, fmt.Errorf("error reading the device state file %q: %s", deviceStateFilePath, err)
+	}
+
+	if string(deviceStateFileContent) != fmt.Sprintf("%s\n", runningState) {
+		return false, nil
+	}
+
+	return true, nil
+}
+
 func (m *mounter) IsFormatted(source string) (bool, error) {
 	if source == "" {
 		return false, errors.New("source is not specified")

driver/mounter_test.go

@@ -0,0 +1,110 @@
+package driver
+
+import (
+	"errors"
+	"fmt"
+	"testing"
+
+	"github.com/sirupsen/logrus"
+)
+
+type testAttachmentValidator struct {
+	readFileFunc     func(name string) ([]byte, error)
+	evalSymlinksFunc func(path string) (string, error)
+}
+
+func (av *testAttachmentValidator) readFile(name string) ([]byte, error) {
+	return av.readFileFunc(name)
+}
+
+func (av *testAttachmentValidator) evalSymlinks(path string) (string, error) {
+	return av.evalSymlinksFunc(path)
+}
+
+func Test_mounter_IsAttached(t *testing.T) {
+	t.Run("could not evaluate the symbolic link", func(t *testing.T) {
+		m := &mounter{
+			log:      logrus.NewEntry(logrus.New()),
+			kMounter: nil,
+			attachmentValidator: &testAttachmentValidator{
+				readFileFunc: nil,
+				evalSymlinksFunc: func(path string) (string, error) {
+					return "", errors.New("error")
+				},
+			},
+		}
+
+		_, err := m.IsAttached("")
+		if err == nil {
+			t.Error("expected error")
+		}
+	})
+
+	t.Run("error reading the device state file", func(t *testing.T) {
+		m := &mounter{
+			log:      logrus.NewEntry(logrus.New()),
+			kMounter: nil,
+			attachmentValidator: &testAttachmentValidator{
+				readFileFunc: func(name string) ([]byte, error) {
+					return nil, errors.New("error")
+				},
+				evalSymlinksFunc: func(path string) (string, error) {
+					return "/dev/sda", nil
+				},
+			},
+		}
+
+		_, err := m.IsAttached("")
+		if err == nil {
+			t.Error("expected error")
+		}
+	})
+
+	t.Run("state file content does not indicate a running state", func(t *testing.T) {
+		m := &mounter{
+			log:      logrus.NewEntry(logrus.New()),
+			kMounter: nil,
+			attachmentValidator: &testAttachmentValidator{
+				readFileFunc: func(name string) ([]byte, error) {
+					return []byte("not-running\n"), nil
+				},
+				evalSymlinksFunc: func(path string) (string, error) {
+					return "/dev/sda", nil
+				},
+			},
+		}
+
+		isAttached, err := m.IsAttached("")
+		if err != nil {
+			t.Errorf("expected no error: %s", err)
+		}
+
+		if isAttached {
+			t.Errorf("IsAttached() must return false when the state file does not contains the content: %s\n", runningState)
+		}
+	})
+
+	t.Run("state file content indicates a running state", func(t *testing.T) {
+		m := &mounter{
+			log:      logrus.NewEntry(logrus.New()),
+			kMounter: nil,
+			attachmentValidator: &testAttachmentValidator{
+				readFileFunc: func(name string) ([]byte, error) {
+					return []byte(fmt.Sprintf("%s\n", runningState)), nil
+				},
+				evalSymlinksFunc: func(path string) (string, error) {
+					return "/dev/sda", nil
+				},
+			},
+		}
+
+		isAttached, err := m.IsAttached("")
+		if err != nil {
+			t.Errorf("expected no error: %s", err)
+		}
+
+		if !isAttached {
+			t.Errorf("IsAttached() must return true when the state file does contains the content: %s\n", runningState)
+		}
+	})
+}