Merge pull request minio#16 from Praveenrajmani/add-sync

Framework to match the probed device with the list of drives

pkg/sys/sys_test.go

@@ -99,83 +99,3 @@ func TestGetBlockFile(t1 *testing.T) {
 	}
 
 }
-
-func TestGetRootBlockFile(t1 *testing.T) {
-
-	testCases := []struct {
-		name     string
-		devName  string
-		rootFile string
-	}{
-		{
-			name:     "test1",
-			devName:  "/dev/xvdb",
-			rootFile: "/dev/xvdb",
-		},
-		{
-			name:     "test2",
-			devName:  "/dev/xvdb1",
-			rootFile: "/dev/xvdb1",
-		},
-		{
-			name:     "test3",
-			devName:  "/var/lib/direct-csi/devices/xvdb",
-			rootFile: "/dev/xvdb",
-		},
-		{
-			name:     "test4",
-			devName:  "/var/lib/direct-csi/devices/xvdb-part-3",
-			rootFile: "/dev/xvdb3",
-		},
-		{
-			name:     "test5",
-			devName:  "/var/lib/direct-csi/devices/xvdb-part-15",
-			rootFile: "/dev/xvdb15",
-		},
-		{
-			name:     "test6",
-			devName:  "/var/lib/direct-csi/devices/nvmen1p-part-4",
-			rootFile: "/dev/nvmen1p4",
-		},
-		{
-			name:     "test7",
-			devName:  "/var/lib/direct-csi/devices/nvmen12p-part-11",
-			rootFile: "/dev/nvmen12p11",
-		},
-		{
-			name:     "test8",
-			devName:  "/var/lib/direct-csi/devices/loop0",
-			rootFile: "/dev/loop0",
-		},
-		{
-			name:     "test9",
-			devName:  "/var/lib/direct-csi/devices/loop-part-5",
-			rootFile: "/dev/loop5",
-		},
-		{
-			name:     "test10",
-			devName:  "/var/lib/direct-csi/devices/loop-part-12",
-			rootFile: "/dev/loop12",
-		},
-		{
-			name:     "test11",
-			devName:  "loop12",
-			rootFile: "/dev/loop12",
-		},
-		{
-			name:     "test12",
-			devName:  "loop0",
-			rootFile: "/dev/loop0",
-		},
-	}
-
-	for _, tt := range testCases {
-		t1.Run(tt.name, func(t1 *testing.T) {
-			rootFile := getRootBlockFile(tt.devName)
-			if rootFile != tt.rootFile {
-				t1.Errorf("Test case name %s: Expected root file = (%s) got: %s", tt.name, tt.rootFile, rootFile)
-			}
-		})
-	}
-
-}

pkg/sys/sys_utils.go

@@ -50,21 +50,6 @@ func makeBlockDeviceName(devName string) string {
 	return strings.Join([]string{dName, partNumStr}, DirectCSIPartitionInfix)
 }
 
-func getRootBlockFile(devName string) string {
-	if strings.Contains(devName, DirectCSIDevRoot) {
-		return getRootBlockFile(filepath.Base(devName))
-	}
-	if strings.HasPrefix(devName, HostDevRoot) {
-		return devName
-	}
-	return filepath.Join(HostDevRoot, makeRootDeviceName(devName))
-}
-
-func makeRootDeviceName(devName string) string {
-	cleanPrefix := strings.Replace(devName, DirectCSIPartitionInfix, "", 1)
-	return strings.ReplaceAll(cleanPrefix, DirectCSIPartitionInfix, HostPartitionInfix)
-}
-
 func splitDevAndPartNum(s string) (string, int) {
 	possibleNum := strings.Builder{}
 	toRet := strings.Builder{}

pkg/uevent/indexer.go

@@ -29,7 +29,6 @@ import (
 
 var (
 	errNotDirectCSIDriveObject = errors.New("not a directcsidrive object")
-	errNoMatchFound            = errors.New("no matching drive found")
 )
 
 type indexer struct {
@@ -120,15 +119,6 @@ func (i *indexer) validateDevice(device *sys.Device) (bool, error) {
 	return true, nil
 }
 
-func (i *indexer) getMatchingDrive(device *sys.Device) (*directcsi.DirectCSIDrive, error) {
-	filteredDrives, err := i.filterDrivesByUEventFSUUID(device.UeventFSUUID)
-	if err != nil {
-		return nil, err
-	}
-	// To-Do/Fix-me: run matching algorithm to find matching drive
-	return filteredDrives[0], errNoMatchFound
-}
-
 func (i *indexer) filterDrivesByUEventFSUUID(fsuuid string) ([]*directcsi.DirectCSIDrive, error) {
 	objects := i.store.List()
 	filteredDrives := []*directcsi.DirectCSIDrive{}
@@ -147,3 +137,19 @@ func (i *indexer) filterDrivesByUEventFSUUID(fsuuid string) ([]*directcsi.Direct
 	}
 	return filteredDrives, nil
 }
+
+func (i *indexer) listDrives() ([]*directcsi.DirectCSIDrive, error) {
+	objects := i.store.List()
+	drives := []*directcsi.DirectCSIDrive{}
+	for _, obj := range objects {
+		directCSIDrive, ok := obj.(*directcsi.DirectCSIDrive)
+		if !ok {
+			return nil, errNotDirectCSIDriveObject
+		}
+		if directCSIDrive.Status.NodeName != i.nodeID {
+			continue
+		}
+		drives = append(drives, directCSIDrive)
+	}
+	return drives, nil
+}

pkg/uevent/listener.go

@@ -41,6 +41,8 @@ const (
 	Add    action = "add"
 	Change action = "change"
 	Remove action = "remove"
+	// internal
+	Sync action = "sync"
 )
 
 var (
@@ -190,46 +192,74 @@ func (l *listener) handle(ctx context.Context, dEvent *deviceEvent) error {
 
 	if err := device.ProbeHostInfo(); err != nil {
 		// if drive is deleted
+		// FIXME: should we ignore errNotExist event for update, add and sync?
 		if !errors.Is(err, os.ErrNotExist) {
 			return err
 		}
 	}
 
+	drives, err := l.indexer.listDrives()
+	if err != nil {
+		return err
+	}
+	drive, matchResult := runMatchers(drives, device, stageOneMatchers, stageTwoMatchers)
+
 	switch dEvent.action {
-	case Add, Change:
-		return l.processUpdate(ctx, device)
+	case Add:
+		return l.processAdd(ctx, matchResult, device, drive)
+	case Change, Sync:
+		return l.processUpdate(ctx, matchResult, device, drive)
 	case Remove:
-		return l.processRemove(ctx, device)
+		return l.processRemove(ctx, matchResult, device, drive)
 	default:
 		return fmt.Errorf("invalid device action: %s", dEvent.action)
 	}
 }
 
-func (l *listener) processUpdate(ctx context.Context, device *sys.Device) error {
-	drive, err := l.indexer.getMatchingDrive(device)
-	switch {
-	case errors.Is(err, errNoMatchFound):
+func (l *listener) processAdd(ctx context.Context, matchResult matchResult, device *sys.Device, drive *directcsi.DirectCSIDrive) error {
+	switch matchResult {
+	case noMatch:
+		return l.handler.Add(ctx, device)
+	case changed, noChange:
+		klog.V(3).Infof("ignoring ADD action for the device %s as the corresponding drive match %s is found", device.DevPath(), drive.Name)
+		return nil
+	case tooManyMatches:
+		klog.V(3).Infof("ignoring ADD action as too many matches are found for the device %s", device.DevPath())
+		return nil
+	default:
+		return fmt.Errorf("invalid match result: %v", matchResult)
+	}
+}
+
+func (l *listener) processUpdate(ctx context.Context, matchResult matchResult, device *sys.Device, drive *directcsi.DirectCSIDrive) error {
+	switch matchResult {
+	case noMatch:
 		return l.handler.Add(ctx, device)
-	case err == nil:
+	case changed:
 		return l.handler.Change(ctx, device, drive)
+	case noChange:
+		return nil
+	case tooManyMatches:
+		return fmt.Errorf("too many matches found for device %s while processing UPDATE", device.DevPath())
 	default:
-		return err
+		return fmt.Errorf("invalid match result: %v", matchResult)
 	}
 }
 
-func (l *listener) processRemove(ctx context.Context, device *sys.Device) error {
-	drive, err := l.indexer.getMatchingDrive(device)
-	switch {
-	case errors.Is(err, errNoMatchFound):
+func (l *listener) processRemove(ctx context.Context, matchResult matchResult, device *sys.Device, drive *directcsi.DirectCSIDrive) error {
+	switch matchResult {
+	case noMatch:
 		klog.V(3).InfoS(
 			"matching drive not found",
 			"ACTION", Remove,
 			"DEVICE", device.Name)
 		return nil
-	case err == nil:
+	case changed, noChange:
 		return l.handler.Remove(ctx, device, drive)
+	case tooManyMatches:
+		return fmt.Errorf("too many matches found for device %s while processing DELETE", device.DevPath())
 	default:
-		return err
+		return fmt.Errorf("invalid match result: %v", matchResult)
 	}
 }
 
@@ -253,7 +283,7 @@ func (l *listener) getNextDeviceUEvent(ctx context.Context) (*deviceEvent, error
 
 func (dEvent *deviceEvent) collectUDevData() error {
 	switch dEvent.action {
-	case Add, Change:
+	case Add, Change, Sync:
 		// Older kernels like in CentOS 7 does not send all information about the device,
 		// hence read relevant data from /run/udev/data/b<major>:<minor>
 		runUdevDataMap, err := sys.ReadRunUdevDataByMajorMinor(dEvent.udevData.Major, dEvent.udevData.Minor)
@@ -300,19 +330,6 @@ func (dEvent *deviceEvent) fillMissingUdevData(runUdevData *sys.UDevData) error
 		return errValueMismatch(dEvent.udevData.Path, "partitionnum", dEvent.udevData.Partition, runUdevData.Partition)
 	}
 
-	// Alternate pattern :-
-	//
-	// if runUdevData.WWID != "" {
-	// 	switch dEvent.udevData.WWID {
-	// 	case "":
-	// 		dEvent.udevData.WWID = runUdevData.WWID
-	// 	case runUdevData.WWID:
-	// 	default:
-	// 		errValueMismatch(dEvent.udevData.WWID, "WWID", dEvent.udevData.WWID, runUdevData.WWID)
-	// 	}
-	// }
-	//
-
 	if runUdevData.WWID != "" {
 		if dEvent.udevData.WWID == "" {
 			dEvent.udevData.WWID = runUdevData.WWID
@@ -437,15 +454,11 @@ func (l *listener) sync() error {
 
 		event := &deviceEvent{
 			created:  time.Now().UTC(),
-			action:   Change,
+			action:   Sync,
 			udevData: runUdevData,
 			devPath:  runUdevData.Path,
 		}
 
-		if err = event.fillMissingUdevData(runUdevData); err != nil {
-			return err
-		}
-
 		l.eventQueue.push(event)
 	}