periodically update folder status in pgsql

Author: lovehunter9

pkg/background_task/background_task.go

@@ -79,5 +79,12 @@ func InitBackgroundTaskManager(ctx context.Context) {
 		interval: 0,
 	})
 
+	manager.RegisterTask(Task{
+		name:     "CheckAndUpdateStatus",
+		taskFunc: postgres.CheckAndUpdateStatus,
+		taskType: PeriodicTask,
+		interval: 1 * time.Minute,
+	})
+
 	manager.Start(ctx)
 }

pkg/drives/drive.go

@@ -226,11 +226,11 @@ func (rs *DriveResourceService) GeneratePathList(db *gorm.DB, processor PathProc
 func (rs *DriveResourceService) parsePathToURI(path string) (string, string) {
 	pathSplit := strings.Split(strings.TrimPrefix(path, "/"), "/")
 	if len(pathSplit) < 2 {
-		return "Unknown", path
+		return "unknown", path
 	}
 	if strings.HasPrefix(pathSplit[1], "pvc-userspace-") {
 		if len(pathSplit) == 2 {
-			return "Unknown", path
+			return "unknown", path
 		}
 		if pathSplit[2] == "Data" {
 			return "data", filepath.Join(pathSplit[1:]...)
@@ -240,9 +240,9 @@ func (rs *DriveResourceService) parsePathToURI(path string) (string, string) {
 	}
 	if pathSplit[1] == "External" {
 		externalPath := ParseExternalPath(filepath.Join(pathSplit[2:]...))
-		return "External", externalPath
+		return "external", externalPath
 	}
-	return "Error", path
+	return "error", path
 }
 
 func generateListingData(listing *files.Listing, stopChan <-chan struct{}, dataChan chan<- string, d *common.Data, mountedData []files.DiskInfo) {
@@ -502,14 +502,14 @@ func ParseExternalPath(path string) string {
 		if strings.HasPrefix(path, datum.Path) {
 			idSerial := datum.IDSerial
 			if idSerial == "" {
-				idSerial = "root"
+				idSerial = datum.Type + "_device"
 			}
 			partationUUID := datum.PartitionUUID
 			if partationUUID == "" {
-				partationUUID = "root"
+				partationUUID = datum.Type + "_partition"
 			}
 			return filepath.Join(datum.Type, idSerial, partationUUID, path)
 		}
 	}
-	return "unknown_type/unknown_device/unknow_partation/" + path
+	return ""
 }

pkg/postgres/path_list.go

@@ -8,6 +8,9 @@ import (
 	"gorm.io/gorm"
 	"gorm.io/gorm/clause"
 	"k8s.io/klog/v2"
+	"os"
+	"path/filepath"
+	"strings"
 	"sync"
 	"time"
 )
@@ -22,6 +25,8 @@ type PathList struct {
 	UpdateTime time.Time `gorm:"not null;type:timestamptz;autoUpdateTime"`
 }
 
+var pathListInited bool = false
+
 func createPathListTable() {
 	// Migrate the schema, create the table if it does not exist
 	err := DBServer.AutoMigrate(&PathList{})
@@ -60,6 +65,8 @@ func InitDrivePathList() {
 		}
 	}
 
+	pathListInited = true
+
 	if err := logPathList(); err != nil {
 		fmt.Println("Error logging path list:", err)
 	}
@@ -82,14 +89,31 @@ func GenerateOtherPathList(ctx context.Context) {
 					//klog.Info("~~~Temp log: cookie hasn't come")
 				}
 				mu.Unlock()
+
+				select {
+				case <-ctx.Done():
+					return
+				default:
+				}
 			}
 		}()
 
 		for {
 			mu.Lock()
 			for len(common.BflCookieCache) == 0 {
 				//klog.Info("~~~Temp log: waiting for cookie")
-				cond.Wait()
+				done := ctx.Done()
+				if done != nil {
+					select {
+					case <-done:
+						mu.Unlock()
+						return
+					default:
+						cond.Wait()
+					}
+				} else {
+					cond.Wait()
+				}
 			}
 
 			var srcTypeList = []string{
@@ -136,7 +160,7 @@ func logPathList() error {
 }
 
 func ProcessDirectory(db *gorm.DB, drive, path string, modTime time.Time) error {
-	if drive == "Unknown" || drive == "Error" || path == "" {
+	if drive == "unknown" || drive == "error" || path == "" {
 		// won't deal with these on purpose
 		return nil
 	}
@@ -187,3 +211,70 @@ func batchUpdate(paths []PathList) error {
 		}),
 	}).Create(&paths).Error
 }
+
+func CheckAndUpdateStatus(ctx context.Context) {
+	if !pathListInited {
+		return
+	}
+
+	var pathEntries []PathList
+
+	if err := DBServer.WithContext(ctx).Where("drive IN (?, ?, ?, ?)", "drive", "data", "cache", "external").Find(&pathEntries).Error; err != nil {
+		klog.Errorf("failed to query drive path list: %v", err)
+		return
+	}
+
+	for _, entry := range pathEntries {
+		var fullPath string
+		var exists bool
+		var err error
+
+		switch entry.Drive {
+		case "drive":
+			fullPath = "/data/" + entry.Path
+		case "data":
+			fullPath = "/data/" + entry.Path
+		case "cache":
+			fullPath = "/appcache/" + entry.Path
+		case "external":
+			pathSplit := strings.Split(entry.Path, "/")
+			fullPath = "/data/External/" + filepath.Join(pathSplit[3:]...)
+		default:
+			continue
+		}
+
+		exists, err = pathExists(fullPath)
+		if err != nil {
+			klog.Errorf("failed to check if path exists: %v", err)
+			return
+		}
+
+		var newStatus int
+		if exists {
+			newStatus = 0
+		} else {
+			newStatus = 1
+		}
+
+		if entry.Status == newStatus {
+			continue
+		}
+
+		if err := DBServer.WithContext(ctx).Model(&PathList{}).Where("drive = ? AND path = ?", entry.Drive, entry.Path).Update("status", newStatus).Error; err != nil {
+			klog.Errorf("failed to update drive path status: %v", err)
+			return
+		}
+	}
+
+	return
+}
+
+func pathExists(path string) (bool, error) {
+	_, err := os.Stat(path)
+	if os.IsNotExist(err) {
+		return false, nil
+	} else if err != nil {
+		return false, err
+	}
+	return true, nil
+}