feat: add disk bandwidth information for host

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

go.mod

@@ -3,7 +3,7 @@ module d7y.io/dragonfly/v2
 go 1.23.0
 
 require (
-	d7y.io/api/v2 v2.0.169
+	d7y.io/api/v2 v2.0.171
 	github.com/MysteriousPotato/go-lockable v1.0.0
 	github.com/RichardKnop/machinery v1.10.8
 	github.com/Showmax/go-fqdn v1.0.0

go.sum

@@ -53,8 +53,8 @@ cloud.google.com/go/storage v1.5.0/go.mod h1:tpKbwo567HUNpVclU5sGELwQWBDZ8gh0Zeo
 cloud.google.com/go/storage v1.6.0/go.mod h1:N7U0C8pVQ/+NIKOBQyamJIeKQKkZ+mxpohlUTyfDhBk=
 cloud.google.com/go/storage v1.8.0/go.mod h1:Wv1Oy7z6Yz3DshWRJFhqM/UCfaWIRTdp0RXyy7KQOVs=
 cloud.google.com/go/storage v1.10.0/go.mod h1:FLPqc6j+Ki4BU591ie1oL6qBQGu2Bl/tZ9ullr3+Kg0=
-d7y.io/api/v2 v2.0.169 h1:CKxPnhXJ0FNOtyATZ5pw5yolRhV6mhlFnEOvgBs9cRA=
-d7y.io/api/v2 v2.0.169/go.mod h1:s3ovYyCQQ9RHUC+RMpAZYI075vkaz/PcLpoyTZqvvOg=
+d7y.io/api/v2 v2.0.171 h1:iHMAhim/BFJ6MhZzsGMmVqF/h0Atw59g/0GuYFyhGxg=
+d7y.io/api/v2 v2.0.171/go.mod h1:HLM5CjwBmy1pDGNUsUNkQeQPItblnHmeTJBEvBDbGnE=
 dmitri.shuralyov.com/gpu/mtl v0.0.0-20190408044501-666a987793e9/go.mod h1:H6x//7gZCb22OMCxBHrMx7a5I7Hp++hsVxbQ4BYO7hU=
 github.com/Azure/azure-sdk-for-go v16.2.1+incompatible/go.mod h1:9XXNKU+eRnpl9moKnB4QOLf1HestfXbmab5FXxiDBjc=
 github.com/Azure/azure-sdk-for-go/sdk/azcore v1.0.0/go.mod h1:uGG2W01BaETf0Ozp+QxxKJdMBNRWPdstHG0Fmdwn1/U=

scheduler/announcer/announcer.go

@@ -26,7 +26,6 @@ import (
 	logger "d7y.io/dragonfly/v2/internal/dflog"
 	managerclient "d7y.io/dragonfly/v2/pkg/rpc/manager/client"
 	"d7y.io/dragonfly/v2/scheduler/config"
-	"d7y.io/dragonfly/v2/scheduler/storage"
 )
 
 // Announcer is the interface used for announce service.
@@ -42,19 +41,17 @@ type Announcer interface {
 type announcer struct {
 	config        *config.Config
 	managerClient managerclient.V2
-	storage       storage.Storage
 	done          chan struct{}
 }
 
 // Option is a functional option for configuring the announcer.
 type Option func(s *announcer)
 
 // New returns a new Announcer interface.
-func New(cfg *config.Config, managerClient managerclient.V2, storage storage.Storage, schedulerFeatures []string, options ...Option) (Announcer, error) {
+func New(cfg *config.Config, managerClient managerclient.V2, schedulerFeatures []string, options ...Option) (Announcer, error) {
 	a := &announcer{
 		config:        cfg,
 		managerClient: managerClient,
-		storage:       storage,
 		done:          make(chan struct{}),
 	}
 

scheduler/announcer/announcer_test.go

@@ -30,7 +30,6 @@ import (
 	managertypes "d7y.io/dragonfly/v2/manager/types"
 	managerclientmocks "d7y.io/dragonfly/v2/pkg/rpc/manager/client/mocks"
 	"d7y.io/dragonfly/v2/scheduler/config"
-	storagemocks "d7y.io/dragonfly/v2/scheduler/storage/mocks"
 )
 
 var (
@@ -116,10 +115,9 @@ func TestAnnouncer_New(t *testing.T) {
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
 			mockManagerClient := managerclientmocks.NewMockV2(ctl)
-			mockStorage := storagemocks.NewMockStorage(ctl)
 			tc.mock(mockManagerClient.EXPECT())
 
-			a, err := New(tc.config, mockManagerClient, mockStorage, managertypes.DefaultSchedulerFeatures)
+			a, err := New(tc.config, mockManagerClient, managertypes.DefaultSchedulerFeatures)
 			tc.expect(t, a, err)
 		})
 	}
@@ -134,7 +132,7 @@ func TestAnnouncer_Serve(t *testing.T) {
 		config *config.Config
 		data   []byte
 		sleep  func()
-		mock   func(data []byte, m *managerclientmocks.MockV2MockRecorder, ms *storagemocks.MockStorageMockRecorder)
+		mock   func(data []byte, m *managerclientmocks.MockV2MockRecorder)
 		except func(t *testing.T, a Announcer)
 	}{
 		{
@@ -161,7 +159,7 @@ func TestAnnouncer_Serve(t *testing.T) {
 			sleep: func() {
 				time.Sleep(3 * time.Second)
 			},
-			mock: func(data []byte, m *managerclientmocks.MockV2MockRecorder, ms *storagemocks.MockStorageMockRecorder) {
+			mock: func(data []byte, m *managerclientmocks.MockV2MockRecorder) {
 				gomock.InOrder(
 					m.UpdateScheduler(gomock.Any(), gomock.Eq(&managerv2.UpdateSchedulerRequest{
 						SourceType:         managerv2.SourceType_SCHEDULER_SOURCE,
@@ -190,10 +188,9 @@ func TestAnnouncer_Serve(t *testing.T) {
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
 			mockManagerClient := managerclientmocks.NewMockV2(ctl)
-			mockStorage := storagemocks.NewMockStorage(ctl)
 
-			tc.mock(tc.data, mockManagerClient.EXPECT(), mockStorage.EXPECT())
-			a, err := New(tc.config, mockManagerClient, mockStorage, managertypes.DefaultSchedulerFeatures)
+			tc.mock(tc.data, mockManagerClient.EXPECT())
+			a, err := New(tc.config, mockManagerClient, managertypes.DefaultSchedulerFeatures)
 			if err != nil {
 				t.Fatal(err)
 			}
@@ -267,10 +264,9 @@ func TestAnnouncer_announceToManager(t *testing.T) {
 			ctl := gomock.NewController(t)
 			defer ctl.Finish()
 			mockManagerClient := managerclientmocks.NewMockV2(ctl)
-			mockStorage := storagemocks.NewMockStorage(ctl)
 			tc.mock(mockManagerClient.EXPECT())
 
-			a, err := New(tc.config, mockManagerClient, mockStorage, managertypes.DefaultSchedulerFeatures)
+			a, err := New(tc.config, mockManagerClient, managertypes.DefaultSchedulerFeatures)
 			if err != nil {
 				t.Fatal(err)
 			}

scheduler/resource/persistentcache/host.go

@@ -116,145 +116,151 @@ type Host struct {
 // CPU contains content for cpu.
 type CPU struct {
 	// Number of logical cores in the system.
-	LogicalCount uint32 `csv:"logicalCount"`
+	LogicalCount uint32
 
 	// Number of physical cores in the system.
-	PhysicalCount uint32 `csv:"physicalCount"`
+	PhysicalCount uint32
 
 	// Percent calculates the percentage of cpu used.
-	Percent float64 `csv:"percent"`
+	Percent float64
 
 	// Calculates the percentage of cpu used by process.
-	ProcessPercent float64 `csv:"processPercent"`
+	ProcessPercent float64
 
 	// Times contains the amounts of time the CPU has spent performing different kinds of work.
-	Times CPUTimes `csv:"times"`
+	Times CPUTimes
 }
 
 // CPUTimes contains content for cpu times.
 type CPUTimes struct {
 	// CPU time of user.
-	User float64 `csv:"user"`
+	User float64
 
 	// CPU time of system.
-	System float64 `csv:"system"`
+	System float64
 
 	// CPU time of idle.
-	Idle float64 `csv:"idle"`
+	Idle float64
 
 	// CPU time of nice.
-	Nice float64 `csv:"nice"`
+	Nice float64
 
 	// CPU time of iowait.
-	Iowait float64 `csv:"iowait"`
+	Iowait float64
 
 	// CPU time of irq.
-	Irq float64 `csv:"irq"`
+	Irq float64
 
 	// CPU time of softirq.
-	Softirq float64 `csv:"softirq"`
+	Softirq float64
 
 	// CPU time of steal.
-	Steal float64 `csv:"steal"`
+	Steal float64
 
 	// CPU time of guest.
-	Guest float64 `csv:"guest"`
+	Guest float64
 
 	// CPU time of guest nice.
-	GuestNice float64 `csv:"guestNice"`
+	GuestNice float64
 }
 
 // Memory contains content for memory.
 type Memory struct {
 	// Total amount of RAM on this system.
-	Total uint64 `csv:"total"`
+	Total uint64
 
 	// RAM available for programs to allocate.
-	Available uint64 `csv:"available"`
+	Available uint64
 
 	// RAM used by programs.
-	Used uint64 `csv:"used"`
+	Used uint64
 
 	// Percentage of RAM used by programs.
-	UsedPercent float64 `csv:"usedPercent"`
+	UsedPercent float64
 
 	// Calculates the percentage of memory used by process.
-	ProcessUsedPercent float64 `csv:"processUsedPercent"`
+	ProcessUsedPercent float64
 
 	// This is the kernel's notion of free memory.
-	Free uint64 `csv:"free"`
+	Free uint64
 }
 
 // Network contains content for network.
 type Network struct {
 	// Return count of tcp connections opened and status is ESTABLISHED.
-	TCPConnectionCount uint32 `csv:"tcpConnectionCount"`
+	TCPConnectionCount uint32
 
 	// Return count of upload tcp connections opened and status is ESTABLISHED.
-	UploadTCPConnectionCount uint32 `csv:"uploadTCPConnectionCount"`
+	UploadTCPConnectionCount uint32
 
 	// Location path(area|country|province|city|...).
-	Location string `csv:"location"`
+	Location string
 
 	// IDC where the peer host is located
-	IDC string `csv:"idc"`
+	IDC string
 
 	// Download rate of the host, unit is byte/s.
-	DownloadRate uint64 `csv:"downloadRate"`
+	DownloadRate uint64
 
 	// Download rate limit of the host, unit is byte/s.
-	DownloadRateLimit uint64 `csv:"downloadRateLimit"`
+	DownloadRateLimit uint64
 
 	// Upload rate of the host, unit is byte/s.
-	UploadRate uint64 `csv:"uploadRate"`
+	UploadRate uint64
 
 	// Upload rate limit of the host, unit is byte/s.
-	UploadRateLimit uint64 `csv:"uploadRateLimit"`
+	UploadRateLimit uint64
 }
 
 // Build contains content for build.
 type Build struct {
 	// Git version.
-	GitVersion string `csv:"gitVersion"`
+	GitVersion string
 
 	// Git commit.
-	GitCommit string `csv:"gitCommit"`
+	GitCommit string
 
 	// Golang version.
-	GoVersion string `csv:"goVersion"`
+	GoVersion string
 
 	// Rust version.
-	RustVersion string `csv:"rustVersion"`
+	RustVersion string
 
 	// Build platform.
-	Platform string `csv:"platform"`
+	Platform string
 }
 
 // Disk contains content for disk.
 type Disk struct {
 	// Total amount of disk on the data path of dragonfly.
-	Total uint64 `csv:"total"`
+	Total uint64
 
 	// Free amount of disk on the data path of dragonfly.
-	Free uint64 `csv:"free"`
+	Free uint64
 
 	// Used amount of disk on the data path of dragonfly.
-	Used uint64 `csv:"used"`
+	Used uint64
 
 	// Used percent of disk on the data path of dragonfly directory.
-	UsedPercent float64 `csv:"usedPercent"`
+	UsedPercent float64
 
 	// Total amount of indoes on the data path of dragonfly directory.
-	InodesTotal uint64 `csv:"inodesTotal"`
+	InodesTotal uint64
 
 	// Used amount of indoes on the data path of dragonfly directory.
-	InodesUsed uint64 `csv:"inodesUsed"`
+	InodesUsed uint64
 
 	// Free amount of indoes on the data path of dragonfly directory.
-	InodesFree uint64 `csv:"inodesFree"`
+	InodesFree uint64
 
 	// Used percent of indoes on the data path of dragonfly directory.
-	InodesUsedPercent float64 `csv:"inodesUsedPercent"`
+	InodesUsedPercent float64
+
+	// Disk write bandwidth, unit is byte/s.
+	WriteBandwidth uint64
+
+	// Disk read bandwidth, unit is byte/s.
+	ReadBandwidth uint64
 }
 
 // New host instance.

scheduler/resource/persistentcache/host_manager.go

@@ -380,6 +380,18 @@ func (h *hostManager) Load(ctx context.Context, hostID string) (*Host, bool) {
 		return nil, false
 	}
 
+	diskWriteBandwidth, err := strconv.ParseUint(rawHost["disk_write_bandwidth"], 10, 64)
+	if err != nil {
+		log.Errorf("parsing disk write bandwidth failed: %v", err)
+		return nil, false
+	}
+
+	diskReadBandwidth, err := strconv.ParseUint(rawHost["disk_read_bandwidth"], 10, 64)
+	if err != nil {
+		log.Errorf("parsing disk read bandwidth failed: %v", err)
+		return nil, false
+	}
+
 	disk := Disk{
 		Total:             diskTotal,
 		Free:              diskFree,
@@ -389,6 +401,8 @@ func (h *hostManager) Load(ctx context.Context, hostID string) (*Host, bool) {
 		InodesUsed:        diskInodesUsed,
 		InodesFree:        diskInodesFree,
 		InodesUsedPercent: diskInodesUsedPercent,
+		WriteBandwidth:    diskWriteBandwidth,
+		ReadBandwidth:     diskReadBandwidth,
 	}
 
 	build := Build{
@@ -498,6 +512,8 @@ func (h *hostManager) Store(ctx context.Context, host *Host) error {
 		"disk_inodes_used", host.Disk.InodesUsed,
 		"disk_inodes_free", host.Disk.InodesFree,
 		"disk_inodes_used_percent", host.Disk.InodesUsedPercent,
+		"disk_write_bandwidth", host.Disk.WriteBandwidth,
+		"disk_read_bandwidth", host.Disk.ReadBandwidth,
 		"build_git_version", host.Build.GitVersion,
 		"build_git_commit", host.Build.GitCommit,
 		"build_go_version", host.Build.GoVersion,