feat: add peer manager for persistent cache task

Signed-off-by: Gaius <gaius.qi@gmail.com>

Author: gaius-qi

internal/dflog/logger.go

@@ -141,15 +141,15 @@ func WithTask(taskID, url string) *SugaredLoggerOnWith {
 	}
 }
 
-func WithPersistentCacheTask(taskID string) *SugaredLoggerOnWith {
+func WithHost(hostID, hostname, ip string) *SugaredLoggerOnWith {
 	return &SugaredLoggerOnWith{
-		withArgs: []any{"taskID", taskID},
+		withArgs: []any{"hostID", hostID, "hostname", hostname, "ip", ip},
 	}
 }
 
-func WithHost(hostID, hostname, ip string) *SugaredLoggerOnWith {
+func WithPeerID(peerID string) *SugaredLoggerOnWith {
 	return &SugaredLoggerOnWith{
-		withArgs: []any{"hostID", hostID, "hostname", hostname, "ip", ip},
+		withArgs: []any{"peerID", peerID},
 	}
 }
 

pkg/redis/redis.go

@@ -141,15 +141,31 @@ func MakePersistentCacheTasksInScheduler(schedulerClusterID uint) string {
 	return MakeKeyInScheduler(SchedulerClustersNamespace, fmt.Sprintf("%d:%s", schedulerClusterID, PersistentCacheTasksNamespace))
 }
 
+// MakePersistentCachePeersOfPersistentCacheTaskInScheduler make persistent cache peers of persistent cache task in scheduler.
+func MakePersistentCachePeersOfPersistentCacheTaskInScheduler(schedulerClusterID uint, taskID string) string {
+	return MakeKeyInScheduler(SchedulerClustersNamespace, fmt.Sprintf("%d:%s:%s:%s", schedulerClusterID, PersistentCacheTasksNamespace, taskID, PersistentCachePeersNamespace))
+}
+
 // MakePersistentCachePeerKeyInScheduler make persistent cache peer key in scheduler.
 func MakePersistentCachePeerKeyInScheduler(schedulerClusterID uint, peerID string) string {
 	return MakeKeyInScheduler(SchedulerClustersNamespace, fmt.Sprintf("%d:%s:%s", schedulerClusterID, PersistentCachePeersNamespace, peerID))
 }
 
+// MakePersistentCachePeersInScheduler make persistent cache peers in scheduler.
 func MakePersistentCachePeersInScheduler(schedulerClusterID uint) string {
 	return MakeKeyInScheduler(SchedulerClustersNamespace, fmt.Sprintf("%d:%s", schedulerClusterID, PersistentCachePeersNamespace))
 }
 
+// MakePersistentCacheHostsOfPersistentCachePeerInScheduler make persistent cache hosts of persistent cache peer in scheduler.
+func MakePersistentCacheHostsOfPersistentCachePeerInScheduler(schedulerClusterID uint, peerID string) string {
+	return MakeKeyInScheduler(SchedulerClustersNamespace, fmt.Sprintf("%d:%s:%s:%s", schedulerClusterID, PersistentCachePeersNamespace, peerID, PersistentCacheHostsNamespace))
+}
+
+// MakePersistentCacheTasksOfPersistentCachePeerInScheduler make persistent cache tasks of persistent cache peer in scheduler.
+func MakePersistentCacheTasksOfPersistentCachePeerInScheduler(schedulerClusterID uint, peerID string) string {
+	return MakeKeyInScheduler(SchedulerClustersNamespace, fmt.Sprintf("%d:%s:%s:%s", schedulerClusterID, PersistentCachePeersNamespace, peerID, PersistentCacheTasksNamespace))
+}
+
 // MakePersistentCacheHostKeyInScheduler make persistent cache host key in scheduler.
 func MakePersistentCacheHostKeyInScheduler(schedulerClusterID uint, hostID string) string {
 	return MakeKeyInScheduler(SchedulerClustersNamespace, fmt.Sprintf("%d:%s:%s", schedulerClusterID, PersistentCacheHostsNamespace, hostID))
@@ -159,3 +175,8 @@ func MakePersistentCacheHostKeyInScheduler(schedulerClusterID uint, hostID strin
 func MakePersistentCacheHostsInScheduler(schedulerClusterID uint) string {
 	return MakeKeyInScheduler(SchedulerClustersNamespace, fmt.Sprintf("%d:%s", schedulerClusterID, PersistentCacheHostsNamespace))
 }
+
+// MakePersistentCachePeersOfPersistentCacheHostInScheduler make persistent cache peers of persistent cache host in scheduler.
+func MakePersistentCachePeersOfPersistentCacheHostInScheduler(schedulerClusterID uint, hostID string) string {
+	return MakeKeyInScheduler(SchedulerClustersNamespace, fmt.Sprintf("%d:%s:%s:%s", schedulerClusterID, PersistentCacheHostsNamespace, hostID, PersistentCachePeersNamespace))
+}

scheduler/resource/persistentcache/host_manager.go

@@ -20,7 +20,6 @@ package persistentcache
 
 import (
 	"context"
-	"fmt"
 	"strconv"
 	"time"
 
@@ -65,138 +64,139 @@ func newHostManager(cfg *config.Config, rdb redis.UniversalClient) HostManager {
 
 // Load returns host for a key.
 func (t *hostManager) Load(ctx context.Context, hostID string) (*Host, bool) {
+	log := logger.WithHostID(hostID)
 	rawHost, err := t.rdb.HGetAll(ctx, pkgredis.MakePersistentCacheHostKeyInScheduler(t.config.Manager.SchedulerClusterID, hostID)).Result()
 	if err != nil {
-		fmt.Println("getting host failed from Redis:", err)
+		log.Errorf("getting host failed from redis:", err)
 		return nil, false
 	}
 
 	// Set integer fields from raw host.
 	port, err := strconv.ParseInt(rawHost["port"], 10, 32)
 	if err != nil {
-		fmt.Println("parsing port failed:", err)
+		log.Errorf("parsing port failed:", err)
 		return nil, false
 	}
 
 	downloadPort, err := strconv.ParseInt(rawHost["download_port"], 10, 32)
 	if err != nil {
-		fmt.Println("parsing download port failed:", err)
+		log.Errorf("parsing download port failed:", err)
 		return nil, false
 	}
 
 	concurrentUploadLimit, err := strconv.ParseInt(rawHost["concurrent_upload_limit"], 10, 32)
 	if err != nil {
-		fmt.Println("parsing concurrent upload limit failed:", err)
+		log.Errorf("parsing concurrent upload limit failed:", err)
 		return nil, false
 	}
 
 	concurrentUploadCount, err := strconv.ParseInt(rawHost["concurrent_upload_count"], 10, 32)
 	if err != nil {
-		fmt.Println("parsing concurrent upload count failed:", err)
+		log.Errorf("parsing concurrent upload count failed:", err)
 		return nil, false
 	}
 
 	uploadCount, err := strconv.ParseInt(rawHost["upload_count"], 10, 64)
 	if err != nil {
-		fmt.Println("parsing upload count failed:", err)
+		log.Errorf("parsing upload count failed:", err)
 		return nil, false
 	}
 
 	uploadFailedCount, err := strconv.ParseInt(rawHost["upload_failed_count"], 10, 64)
 	if err != nil {
-		fmt.Println("parsing upload failed count failed:", err)
+		log.Errorf("parsing upload failed count failed:", err)
 		return nil, false
 	}
 
 	// Set boolean fields from raw host.
 	diableShared, err := strconv.ParseBool(rawHost["disable_shared"])
 	if err != nil {
-		fmt.Println("parsing disable shared failed:", err)
+		log.Errorf("parsing disable shared failed:", err)
 		return nil, false
 	}
 
 	// Set cpu fields from raw host.
 	cpuLogicalCount, err := strconv.ParseUint(rawHost["cpu_logical_count"], 10, 32)
 	if err != nil {
-		fmt.Println("parsing cpu logical count failed:", err)
+		log.Errorf("parsing cpu logical count failed:", err)
 		return nil, false
 	}
 
 	cpuPhysicalCount, err := strconv.ParseUint(rawHost["cpu_physical_count"], 10, 32)
 	if err != nil {
-		fmt.Println("parsing cpu physical count failed:", err)
+		log.Errorf("parsing cpu physical count failed:", err)
 		return nil, false
 	}
 
 	cpuPercent, err := strconv.ParseFloat(rawHost["cpu_percent"], 64)
 	if err != nil {
-		fmt.Println("parsing cpu percent failed:", err)
+		log.Errorf("parsing cpu percent failed:", err)
 		return nil, false
 	}
 
 	cpuProcessPercent, err := strconv.ParseFloat(rawHost["cpu_processe_percent"], 64)
 	if err != nil {
-		fmt.Println("parsing cpu process percent failed:", err)
+		log.Errorf("parsing cpu process percent failed:", err)
 		return nil, false
 	}
 
 	cpuTimesUser, err := strconv.ParseFloat(rawHost["cpu_times_user"], 64)
 	if err != nil {
-		fmt.Println("parsing cpu times user failed:", err)
+		log.Errorf("parsing cpu times user failed:", err)
 		return nil, false
 	}
 
 	cpuTimesSystem, err := strconv.ParseFloat(rawHost["cpu_times_system"], 64)
 	if err != nil {
-		fmt.Println("parsing cpu times system failed:", err)
+		log.Errorf("parsing cpu times system failed:", err)
 		return nil, false
 	}
 
 	cpuTimesIdle, err := strconv.ParseFloat(rawHost["cpu_times_idle"], 64)
 	if err != nil {
-		fmt.Println("parsing cpu times idle failed:", err)
+		log.Errorf("parsing cpu times idle failed:", err)
 		return nil, false
 	}
 
 	cpuTimesNice, err := strconv.ParseFloat(rawHost["cpu_times_nice"], 64)
 	if err != nil {
-		fmt.Println("parsing cpu times nice failed:", err)
+		log.Errorf("parsing cpu times nice failed:", err)
 		return nil, false
 	}
 
 	cpuTimesIowait, err := strconv.ParseFloat(rawHost["cpu_times_iowait"], 64)
 	if err != nil {
-		fmt.Println("parsing cpu times iowait failed:", err)
+		log.Errorf("parsing cpu times iowait failed:", err)
 		return nil, false
 	}
 
 	cpuTimesIrq, err := strconv.ParseFloat(rawHost["cpu_times_irq"], 64)
 	if err != nil {
-		fmt.Println("parsing cpu times irq failed:", err)
+		log.Errorf("parsing cpu times irq failed:", err)
 		return nil, false
 	}
 
 	cpuTimesSoftirq, err := strconv.ParseFloat(rawHost["cpu_times_softirq"], 64)
 	if err != nil {
-		fmt.Println("parsing cpu times softirq failed:", err)
+		log.Errorf("parsing cpu times softirq failed:", err)
 		return nil, false
 	}
 
 	cpuTimesSteal, err := strconv.ParseFloat(rawHost["cpu_times_steal"], 64)
 	if err != nil {
-		fmt.Println("parsing cpu times steal failed:", err)
+		log.Errorf("parsing cpu times steal failed:", err)
 		return nil, false
 	}
 
 	cpuTimesGuest, err := strconv.ParseFloat(rawHost["cpu_times_guest"], 64)
 	if err != nil {
-		fmt.Println("parsing cpu times guest failed:", err)
+		log.Errorf("parsing cpu times guest failed:", err)
 		return nil, false
 	}
 
 	cpuTimesGuestNice, err := strconv.ParseFloat(rawHost["cpu_times_guest_nice"], 64)
 	if err != nil {
-		fmt.Println("parsing cpu times guest nice failed:", err)
+		log.Errorf("parsing cpu times guest nice failed:", err)
 		return nil, false
 	}
 
@@ -222,37 +222,37 @@ func (t *hostManager) Load(ctx context.Context, hostID string) (*Host, bool) {
 	// Set memory fields from raw host.
 	memoryTotal, err := strconv.ParseUint(rawHost["memory_total"], 10, 64)
 	if err != nil {
-		fmt.Println("parsing memory total failed:", err)
+		log.Errorf("parsing memory total failed:", err)
 		return nil, false
 	}
 
 	memoryAvailable, err := strconv.ParseUint(rawHost["memory_available"], 10, 64)
 	if err != nil {
-		fmt.Println("parsing memory available failed:", err)
+		log.Errorf("parsing memory available failed:", err)
 		return nil, false
 	}
 
 	memoryUsed, err := strconv.ParseUint(rawHost["memory_used"], 10, 64)
 	if err != nil {
-		fmt.Println("parsing memory used failed:", err)
+		log.Errorf("parsing memory used failed:", err)
 		return nil, false
 	}
 
 	memoryUsedPercent, err := strconv.ParseFloat(rawHost["memory_used_percent"], 64)
 	if err != nil {
-		fmt.Println("parsing memory used percent failed:", err)
+		log.Errorf("parsing memory used percent failed:", err)
 		return nil, false
 	}
 
 	memoryProcessUsedPercent, err := strconv.ParseFloat(rawHost["memory_processe_used_percent"], 64)
 	if err != nil {
-		fmt.Println("parsing memory process used percent failed:", err)
+		log.Errorf("parsing memory process used percent failed:", err)
 		return nil, false
 	}
 
 	memoryFree, err := strconv.ParseUint(rawHost["memory_free"], 10, 64)
 	if err != nil {
-		fmt.Println("parsing memory free failed:", err)
+		log.Errorf("parsing memory free failed:", err)
 		return nil, false
 	}
 
@@ -268,37 +268,37 @@ func (t *hostManager) Load(ctx context.Context, hostID string) (*Host, bool) {
 	// Set network fields from raw host.
 	networkTCPConnectionCount, err := strconv.ParseUint(rawHost["network_tcp_connection_count"], 10, 32)
 	if err != nil {
-		fmt.Println("parsing network tcp connection count failed:", err)
+		log.Errorf("parsing network tcp connection count failed:", err)
 		return nil, false
 	}
 
 	networkUploadTCPConnectionCount, err := strconv.ParseUint(rawHost["network_upload_tcp_connection_count"], 10, 32)
 	if err != nil {
-		fmt.Println("parsing network upload tcp connection count failed:", err)
+		log.Errorf("parsing network upload tcp connection count failed:", err)
 		return nil, false
 	}
 
 	downloadRate, err := strconv.ParseUint(rawHost["network_download_rate"], 10, 64)
 	if err != nil {
-		fmt.Println("parsing download rate failed:", err)
+		log.Errorf("parsing download rate failed:", err)
 		return nil, false
 	}
 
 	downloadRateLimit, err := strconv.ParseUint(rawHost["network_download_rate_limit"], 10, 64)
 	if err != nil {
-		fmt.Println("parsing download rate limit failed:", err)
+		log.Errorf("parsing download rate limit failed:", err)
 		return nil, false
 	}
 
 	uploadRate, err := strconv.ParseUint(rawHost["network_upload_rate"], 10, 64)
 	if err != nil {
-		fmt.Println("parsing upload rate failed:", err)
+		log.Errorf("parsing upload rate failed:", err)
 		return nil, false
 	}
 
 	uploadRateLimit, err := strconv.ParseUint(rawHost["network_upload_rate_limit"], 10, 64)
 	if err != nil {
-		fmt.Println("parsing upload rate limit failed:", err)
+		log.Errorf("parsing upload rate limit failed:", err)
 		return nil, false
 	}
 
@@ -316,49 +316,49 @@ func (t *hostManager) Load(ctx context.Context, hostID string) (*Host, bool) {
 	// Set disk fields from raw host.
 	diskTotal, err := strconv.ParseUint(rawHost["disk_total"], 10, 64)
 	if err != nil {
-		fmt.Println("parsing disk total failed:", err)
+		log.Errorf("parsing disk total failed:", err)
 		return nil, false
 	}
 
 	diskFree, err := strconv.ParseUint(rawHost["disk_free"], 10, 64)
 	if err != nil {
-		fmt.Println("parsing disk free failed:", err)
+		log.Errorf("parsing disk free failed:", err)
 		return nil, false
 	}
 
 	diskUsed, err := strconv.ParseUint(rawHost["disk_used"], 10, 64)
 	if err != nil {
-		fmt.Println("parsing disk used failed:", err)
+		log.Errorf("parsing disk used failed:", err)
 		return nil, false
 	}
 
 	diskUsedPercent, err := strconv.ParseFloat(rawHost["disk_used_percent"], 64)
 	if err != nil {
-		fmt.Println("parsing disk used percent failed:", err)
+		log.Errorf("parsing disk used percent failed:", err)
 		return nil, false
 	}
 
 	diskInodesTotal, err := strconv.ParseUint(rawHost["disk_inodes_total"], 10, 64)
 	if err != nil {
-		fmt.Println("parsing disk inodes total failed:", err)
+		log.Errorf("parsing disk inodes total failed:", err)
 		return nil, false
 	}
 
 	diskInodesUsed, err := strconv.ParseUint(rawHost["disk_inodes_used"], 10, 64)
 	if err != nil {
-		fmt.Println("parsing disk inodes used failed:", err)
+		log.Errorf("parsing disk inodes used failed:", err)
 		return nil, false
 	}
 
 	diskInodesFree, err := strconv.ParseUint(rawHost["disk_inodes_free"], 10, 64)
 	if err != nil {
-		fmt.Println("parsing disk inodes free failed:", err)
+		log.Errorf("parsing disk inodes free failed:", err)
 		return nil, false
 	}
 
 	diskInodesUsedPercent, err := strconv.ParseFloat(rawHost["disk_inodes_used_percent"], 64)
 	if err != nil {
-		fmt.Println("parsing disk inodes used percent failed:", err)
+		log.Errorf("parsing disk inodes used percent failed:", err)
 		return nil, false
 	}
 
@@ -383,19 +383,19 @@ func (t *hostManager) Load(ctx context.Context, hostID string) (*Host, bool) {
 	// Set time fields from raw host.
 	announceInterval, err := strconv.ParseInt(rawHost["announce_interval"], 10, 32)
 	if err != nil {
-		fmt.Println("parsing announce interval failed:", err)
+		log.Errorf("parsing announce interval failed:", err)
 		return nil, false
 	}
 
 	createdAt, err := time.Parse(time.RFC3339, rawHost["created_at"])
 	if err != nil {
-		fmt.Println("parsing created at failed:", err)
+		log.Errorf("parsing created at failed:", err)
 		return nil, false
 	}
 
 	updatedAt, err := time.Parse(time.RFC3339, rawHost["updated_at"])
 	if err != nil {
-		fmt.Println("parsing updated at failed:", err)
+		log.Errorf("parsing updated at failed:", err)
 		return nil, false
 	}