fix: host ping piggyback storage info (#22962)

Co-authored-by: Qiu Jian <qiujian@yunionyun.com>

pkg/apis/compute/host.go

@@ -460,3 +460,20 @@ type HostAutoMigrateInput struct {
 	AutoMigrateOnHostDown     string `json:"auto_migrate_on_host_down"`
 	AutoMigrateOnHostShutdown string `json:"auto_migrate_on_host_shutdown"`
 }
+
+type SHostStorageStat struct {
+	StorageId string `json:"storage_id"`
+
+	CapacityMb           int64 `json:"capacity_mb"`
+	ActualCapacityUsedMb int64 `json:"actual_capacity_used_mb"`
+}
+
+type SHostPingInput struct {
+	WithData bool `json:"with_data"`
+
+	MemoryUsedMb int `json:"memory_used_mb"`
+
+	RootPartitionUsedCapacityMb int `json:"root_partition_used_capacity_mb"`
+
+	StorageStats []SHostStorageStat `json:"storage_stats"`
+}

pkg/compute/models/hosts.go

@@ -4560,12 +4560,35 @@ func (self *SHost) AllowPerformPing(ctx context.Context,
 	return db.IsAdminAllowPerform(userCred, self, "ping")
 }
 
-func (self *SHost) PerformPing(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data jsonutils.JSONObject) (jsonutils.JSONObject, error) {
+func (self *SHost) PerformPing(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, input api.SHostPingInput) (jsonutils.JSONObject, error) {
 	if self.HostType == api.HOST_TYPE_BAREMETAL {
 		return nil, httperrors.NewNotSupportedError("ping host type %s not support", self.HostType)
 	}
+
+	if input.WithData {
+		// piggyback storage stats info
+		log.Debugf("host ping %s", jsonutils.Marshal(input))
+		for _, si := range input.StorageStats {
+			storageObj, err := StorageManager.FetchById(si.StorageId)
+			if err != nil {
+				log.Errorf("fetch storage %s error %s", si.StorageId, err)
+			} else {
+				storage := storageObj.(*SStorage)
+				_, err := db.Update(storage, func() error {
+					storage.Capacity = si.CapacityMb
+					storage.ActualCapacityUsed = si.ActualCapacityUsedMb
+					return nil
+				})
+				if err != nil {
+					log.Errorf("update storage info error %s", err)
+				}
+			}
+		}
+		self.SetMetadata(ctx, "root_partition_used_capacity_mb", input.RootPartitionUsedCapacityMb, userCred)
+		self.SetMetadata(ctx, "memory_used_mb", input.MemoryUsedMb, userCred)
+	}
 	if self.HostStatus != api.HOST_ONLINE {
-		self.PerformOnline(ctx, userCred, query, data)
+		self.PerformOnline(ctx, userCred, query, nil)
 	} else {
 		self.SaveUpdates(func() error {
 			self.LastPingAt = time.Now()

pkg/hostman/hostinfo/hostinfo.go

@@ -1193,12 +1193,12 @@ func (h *SHostInfo) updateHostMetadata(hostname string) error {
 	return err
 }
 
-func (h *SHostInfo) SyncRootPartitionUsedCapacity() error {
-	data := jsonutils.NewDict()
-	data.Set("root_partition_used_capacity_mb", jsonutils.NewInt(int64(storageman.GetRootPartUsedCapacity())))
-	_, err := modules.Hosts.SetMetadata(h.GetSession(), h.HostId, data)
-	return err
-}
+// func (h *SHostInfo) SyncRootPartitionUsedCapacity() error {
+//	data := jsonutils.NewDict()
+//	data.Set("root_partition_used_capacity_mb", jsonutils.NewInt(int64(storageman.GetRootPartUsedCapacity())))
+//	_, err := modules.Hosts.SetMetadata(h.GetSession(), h.HostId, data)
+//	return err
+// }
 
 func (h *SHostInfo) onUpdateHostInfoSucc(hostbody jsonutils.JSONObject) {
 	h.HostId, _ = hostbody.GetString("id")
@@ -1554,9 +1554,9 @@ func (h *SHostInfo) uploadStorageInfo() {
 			h.onSyncStorageInfoSucc(s, res)
 		}
 	}
-	go storageman.StartSyncStorageSizeTask(
-		time.Duration(options.HostOptions.SyncStorageInfoDurationSecond) * time.Second,
-	)
+	// go storageman.StartSyncStorageSizeTask(
+	//	time.Duration(options.HostOptions.SyncStorageInfoDurationSecond) * time.Second,
+	// )
 	h.probeSyncIsolatedDevicesStep()
 }
 

pkg/hostman/hostinfo/hostpinger.go

@@ -18,9 +18,15 @@ import (
 	"context"
 	"time"
 
+	"github.com/shirou/gopsutil/mem"
+
+	"yunion.io/x/jsonutils"
 	"yunion.io/x/log"
 
+	api "yunion.io/x/onecloud/pkg/apis/compute"
 	"yunion.io/x/onecloud/pkg/hostman/hostutils"
+	"yunion.io/x/onecloud/pkg/hostman/options"
+	"yunion.io/x/onecloud/pkg/hostman/storageman"
 	"yunion.io/x/onecloud/pkg/mcclient"
 	"yunion.io/x/onecloud/pkg/mcclient/modules"
 )
@@ -28,6 +34,8 @@ import (
 type SHostPingTask struct {
 	interval int // second
 	running  bool
+
+	lastStatAt time.Time
 }
 
 type SEndpoint struct {
@@ -56,7 +64,10 @@ func NewHostPingTask(interval int) *SHostPingTask {
 	if interval <= 0 {
 		return nil
 	}
-	return &SHostPingTask{interval, true}
+	return &SHostPingTask{
+		interval: interval,
+		running:  true,
+	}
 }
 
 func (p *SHostPingTask) Start() {
@@ -80,9 +91,31 @@ func (p *SHostPingTask) Start() {
 	}
 }
 
+func (p *SHostPingTask) payload() api.SHostPingInput {
+	data := api.SHostPingInput{}
+
+	now := time.Now()
+	if !p.lastStatAt.IsZero() && now.Before(p.lastStatAt.Add(time.Duration(options.HostOptions.SyncStorageInfoDurationSecond)*time.Second)) {
+		return data
+	}
+
+	p.lastStatAt = now
+	data = storageman.GatherHostStorageStats()
+	data.WithData = true
+	info, err := mem.VirtualMemory()
+	if err != nil {
+		return data
+	}
+	memTotal := int(info.Total / 1024 / 1024)
+	memFree := int(info.Available / 1024 / 1024)
+	memUsed := memTotal - memFree
+	data.MemoryUsedMb = memUsed
+	return data
+}
+
 func (p *SHostPingTask) ping(div int, hostId string) error {
 	res, err := modules.Hosts.PerformAction(hostutils.GetComputeSession(context.Background()),
-		hostId, "ping", nil)
+		hostId, "ping", jsonutils.Marshal(p.payload()))
 	if err != nil {
 		return err
 	} else {

pkg/hostman/hostutils/hostutils.go

@@ -56,7 +56,7 @@ type IHost interface {
 	GetBridgeDev(bridge string) hostbridge.IBridgeDriver
 	GetIsolatedDeviceManager() isolated_device.IsolatedDeviceManager
 
-	SyncRootPartitionUsedCapacity() error
+	// SyncRootPartitionUsedCapacity() error
 
 	GetKubeletConfig() kubelet.KubeletConfig
 }

pkg/hostman/options/options.go

@@ -159,7 +159,7 @@ type SHostOptions struct {
 	HostHealthTimeout    int  `help:"host health timeout" default:"30"`
 	HostLeaseTimeout     int  `help:"lease timeout" default:"10"`
 
-	SyncStorageInfoDurationSecond int  `help:"sync storage size duration, unit is second" default:"60"`
+	SyncStorageInfoDurationSecond int  `help:"sync storage size duration, unit is second, default is every 2 minutes" default:"120"`
 	StartHostIgnoreSysError       bool `help:"start host agent ignore sys error" default:"false"`
 
 	DisableProbeKubelet bool   `help:"Disable probe kubelet config" default:"false"`

pkg/hostman/storageman/core.go

@@ -373,27 +373,44 @@ func CleanRecycleDiskfiles(ctx context.Context, userCred mcclient.TokenCredentia
 	}
 }
 
-func StartSyncStorageSizeTask(interval time.Duration) {
-	log.Infof("Start sync storage size task !!!")
-	for {
-		time.Sleep(interval)
-		manager := GetManager()
-		for i := 0; i < len(manager.Storages); i++ {
-			iS := manager.Storages[i]
-			if iS.StorageType() == api.STORAGE_LOCAL || iS.StorageType() == api.STORAGE_RBD {
-				err := iS.SyncStorageSize()
-				if err != nil {
-					log.Errorf("sync storage %s size failed: %s", iS.GetStorageName(), err)
-				}
-			}
-		}
-		err := manager.host.SyncRootPartitionUsedCapacity()
+func GatherHostStorageStats() api.SHostPingInput {
+	stats := api.SHostPingInput{}
+	stats.RootPartitionUsedCapacityMb = GetRootPartUsedCapacity()
+	manager := GetManager()
+	for i := 0; i < len(manager.Storages); i++ {
+		iS := manager.Storages[i]
+		stat, err := iS.SyncStorageSize()
 		if err != nil {
-			log.Errorf("sync root partition used size failed: %s", err)
+			log.Errorf("sync storage %s size failed: %s", iS.GetStorageName(), err)
+		} else {
+			stat.StorageId = iS.GetId()
+			stats.StorageStats = append(stats.StorageStats, stat)
 		}
 	}
+	return stats
 }
 
+// func StartSyncStorageSizeTask(interval time.Duration) {
+//	log.Infof("Start sync storage size task !!!")
+//	for {
+//		time.Sleep(interval)
+//		manager := GetManager()
+//		for i := 0; i < len(manager.Storages); i++ {
+//			iS := manager.Storages[i]
+//			if iS.StorageType() == api.STORAGE_LOCAL || iS.StorageType() == api.STORAGE_RBD {
+//				err := iS.SyncStorageSize()
+//				if err != nil {
+//					log.Errorf("sync storage %s size failed: %s", iS.GetStorageName(), err)
+//				}
+//			}
+//		}
+//		err := manager.host.SyncRootPartitionUsedCapacity()
+//		if err != nil {
+//			log.Errorf("sync root partition used size failed: %s", err)
+//		}
+//	}
+// }
+
 func GetRootPartTotalCapacity() int {
 	size, err := storageutils.GetTotalSizeMb("/")
 	if err != nil {

pkg/hostman/storageman/storage_base.go

@@ -29,6 +29,7 @@ import (
 	"yunion.io/x/log"
 	"yunion.io/x/pkg/errors"
 
+	api "yunion.io/x/onecloud/pkg/apis/compute"
 	"yunion.io/x/onecloud/pkg/apis/host"
 	"yunion.io/x/onecloud/pkg/cloudcommon/cronman"
 	"yunion.io/x/onecloud/pkg/hostman/hostutils"
@@ -89,7 +90,7 @@ type IStorage interface {
 
 	SetStorageInfo(storageId, storageName string, conf jsonutils.JSONObject) error
 	SyncStorageInfo() (jsonutils.JSONObject, error)
-	SyncStorageSize() error
+	SyncStorageSize() (api.SHostStorageStat, error)
 	StorageType() string
 	GetStorageConf() *jsonutils.JSONDict
 	GetStoragecacheId() string
@@ -271,8 +272,13 @@ func (s *SBaseStorage) SetStorageInfo(storageId, storageName string, conf jsonut
 	return nil
 }
 
-func (s *SBaseStorage) SyncStorageSize() error {
-	return fmt.Errorf("not ipmlement")
+func (s *SBaseStorage) SyncStorageSize() (api.SHostStorageStat, error) {
+	stat := api.SHostStorageStat{
+		StorageId: s.StorageId,
+	}
+	stat.CapacityMb = int64(s.GetCapacity())
+	stat.ActualCapacityUsedMb = int64(s.GetUsedSizeMb())
+	return stat, nil
 }
 
 func (s *SBaseStorage) bindMountTo(sPath string) error {

pkg/hostman/storageman/storage_local.go

@@ -92,14 +92,6 @@ func (s *SLocalStorage) GetComposedName() string {
 	return fmt.Sprintf("host_%s_%s_storage_%d", s.Manager.host.GetMasterIp(), s.StorageType(), s.Index)
 }
 
-func (s *SLocalStorage) SyncStorageSize() error {
-	content := jsonutils.NewDict()
-	content.Set("actual_capacity_used", jsonutils.NewInt(int64(s.GetUsedSizeMb())))
-	_, err := modules.Storages.Put(
-		hostutils.GetComputeSession(context.Background()),
-		s.StorageId, content)
-	return err
-}
 func (s *SLocalStorage) CreateDiskFromBackup(ctx context.Context, disk IDisk, input *SDiskCreateByDiskinfo) error {
 	info := input.DiskInfo
 	backupDir := s.GetBackupDir()

pkg/hostman/storageman/storage_rbd.go

@@ -369,23 +369,23 @@ func (s *SRbdStorage) deleteSnapshot(pool string, diskId string, snapshotId stri
 	return snap.Delete()
 }
 
-func (s *SRbdStorage) SyncStorageSize() error {
-	content := jsonutils.NewDict()
+func (s *SRbdStorage) SyncStorageSize() (api.SHostStorageStat, error) {
+	stat := api.SHostStorageStat{
+		StorageId: s.StorageId,
+	}
+
 	client, err := s.GetClient()
 	if err != nil {
-		return errors.Wrapf(err, "GetClient")
+		return stat, errors.Wrapf(err, "GetClient")
 	}
 	defer client.Close()
 	capacity, err := client.GetCapacity()
 	if err != nil {
-		return errors.Wrapf(err, "GetCapacity")
+		return stat, errors.Wrapf(err, "GetCapacity")
 	}
-	content.Set("capacity", jsonutils.NewInt(int64(capacity.CapacitySizeKb/1024)))
-	content.Set("actual_capacity_used", jsonutils.NewInt(int64(capacity.UsedCapacitySizeKb/1024)))
-	_, err = modules.Storages.Put(
-		hostutils.GetComputeSession(context.Background()),
-		s.StorageId, content)
-	return errors.Wrapf(err, "storage update")
+	stat.CapacityMb = capacity.CapacitySizeKb / 1024
+	stat.ActualCapacityUsedMb = capacity.UsedCapacitySizeKb / 1024
+	return stat, nil
 }
 
 func (s *SRbdStorage) SyncStorageInfo() (jsonutils.JSONObject, error) {

pkg/hostman/storageman/storagehandler/storagehandler.go

@@ -125,10 +125,10 @@ func storageAttach(ctx context.Context, body jsonutils.JSONObject) (interface{},
 	if err := storage.SetStorageInfo(storageId, storageName, storageConf); err != nil {
 		return nil, err
 	}
-	err = storage.SyncStorageSize()
+	/*err = storage.SyncStorageSize()
 	if err != nil {
 		return nil, errors.Wrapf(err, "SyncStorageSize")
-	}
+	}*/
 	resp, err := storage.SyncStorageInfo()
 	if err != nil {
 		return nil, err