opensource/cloudpods
3
0
Fork 0
mirror of https://github.com/yunionio/cloudpods.git synced 2026-09-19 02:37:24 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(region,host): storage add master host (#21472)

Browse Source
This commit is contained in:
wanyaoqi
2024-11-06 10:04:01 +08:00
committed by GitHub
parent 2f7dbbd062
commit 0e7d28b76c
8 changed files with 143 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files
pkg/apis/compute/storage.go
+3 -1
View File
@@ -120,7 +120,6 @@ type StorageCreateInput struct {
CLVMVgName string
// SLVM VG Name
SLVMVgName string
MasterHost string
Lvmlockd bool
}
@@ -183,6 +182,9 @@ type StorageDetails struct {
// 超分比
CommitBound float32 `json:"commit_bound"`
// master host name
MasterHostName string `json:"master_host_name"`
}
func (self StorageDetails) GetMetricTags() map[string]string {
pkg/compute/models/hosts.go
+27 -1
View File
@@ -4644,13 +4644,34 @@ func (hh *SHost) StartSyncAllGuestsStatusTask(ctx context.Context, userCred mccl
}
}
func (hh *SHost) GetStoragesByMasterHost() ([]string, error) {
sq := StorageManager.Query()
sq = sq.In("storage_type", api.SHARED_STORAGE)
sq = sq.Filter(sqlchemy.OR(sqlchemy.Equals(sq.Field("master_host"), hh.Id), sqlchemy.IsNullOrEmpty(sq.Field("master_host"))))
subq := sq.SubQuery()
hsq := HoststorageManager.Query().Equals("host_id", hh.Id)
hsq = hsq.Join(subq, sqlchemy.Equals(subq.Field("id"), hsq.Field("storage_id")))
hostStorages := make([]SHoststorage, 0)
if err := hsq.All(&hostStorages); err != nil && err != sql.ErrNoRows {
return nil, errors.Wrap(err, "get hostStorages")
} else if err == sql.ErrNoRows {
return nil, nil
}
storages := make([]string, len(hostStorages))
for i := range storages {
storages[i] = hostStorages[i].StorageId
}
return storages, nil
}
func (hh *SHost) PerformPing(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, input api.SHostPingInput) (jsonutils.JSONObject, error) {
if hh.HostType == api.HOST_TYPE_BAREMETAL {
return nil, httperrors.NewNotSupportedError("ping host type %s not support", hh.HostType)
}
if input.WithData {
// piggyback storage stats info
log.Debugf("host ping %s", jsonutils.Marshal(input))
log.Debugf("host ping %#v", input)
for _, si := range input.StorageStats {
storageObj, err := StorageManager.FetchById(si.StorageId)
if err != nil {
@@ -4702,6 +4723,11 @@ func (hh *SHost) PerformPing(ctx context.Context, userCred mcclient.TokenCredent
return nil, fmt.Errorf("Get catalog error")
}
result.Set("catalog", catalog)
if storages, err := hh.GetStoragesByMasterHost(); err != nil {
return nil, err
} else {
result.Set("master_host_storages", jsonutils.NewStringArray(storages))
}
appParams := appsrv.AppContextGetParams(ctx)
if appParams != nil {
pkg/compute/models/storages.go
+69 -4
View File
@@ -96,7 +96,7 @@ type SStorage struct {
StoragecacheId string `width:"36" charset:"ascii" nullable:"true" list:"domain" get:"domain" update:"domain" create:"domain_optional"`
// master host id
MasterHost string `width:"36" charset:"ascii" nullable:"true" list:"user" create:"optional" update:"user" json:"master_host"`
MasterHost string `width:"36" charset:"ascii" nullable:"true" list:"user" json:"master_host"`
// indicating whether system disk can be allocated in this storage
// 是否可以用作系统盘存储
@@ -618,6 +618,21 @@ func (manager *SStorageManager) FetchCustomizeColumns(
}
storage := objs[i].(*SStorage)
storageIds[i] = storage.Id
if rows[i].ManagerId == "" && rows[i].MasterHost == "" &&
utils.IsInStringArray(storage.StorageType, api.SHARED_STORAGE) {
if host, err := storage.GetMasterHost(); host != nil {
rows[i].MasterHost = host.Id
rows[i].MasterHostName = host.Name
} else {
log.Errorf("storage %s failed get master host %s", storageIds[i], err)
}
}
if rows[i].MasterHost != "" && rows[i].MasterHostName == "" {
if host := HostManager.FetchHostById(rows[i].MasterHost); host != nil {
rows[i].MasterHostName = host.Name
}
}
rows[i].Capacity = storage.GetCapacity()
rows[i].VCapacity = int64(float32(rows[i].Capacity) * storage.GetOvercommitBound())
rows[i].ActualUsed = storage.ActualCapacityUsed
@@ -750,6 +765,13 @@ func (self *SStorage) GetOvercommitBound() float32 {
}
func (self *SStorage) GetMasterHost() (*SHost, error) {
if self.MasterHost != "" {
host := HostManager.FetchHostById(self.MasterHost)
if host != nil && host.Enabled.IsTrue() && host.HostStatus == api.HOST_ONLINE {
return host, nil
}
}
hosts := HostManager.Query().SubQuery()
hoststorages := HoststorageManager.Query().SubQuery()
@@ -757,18 +779,31 @@ func (self *SStorage) GetMasterHost() (*SHost, error) {
q = q.Filter(sqlchemy.Equals(hoststorages.Field("storage_id"), self.Id))
q = q.IsTrue("enabled")
q = q.Equals("host_status", api.HOST_ONLINE).Asc("id")
if self.MasterHost != "" {
q.Equals("id", self.MasterHost)
}
host := SHost{}
host.SetModelManager(HostManager, &host)
err := q.First(&host)
if err != nil {
return nil, errors.Wrapf(err, "q.First")
}
if utils.IsInStringArray(self.StorageType, api.SHARED_STORAGE) {
if err := self.UpdateMasterHost(host.Id); err != nil {
log.Errorf("storage %s udpate master host failed %s: %s", self.GetName(), host.Id, err)
}
}
return &host, nil
}
func (self *SStorage) UpdateMasterHost(hostId string) error {
_, err := db.Update(self, func() error {
self.MasterHost = hostId
return nil
})
return err
}
func (self *SStorage) GetZoneId() string {
if len(self.ZoneId) > 0 {
return self.ZoneId
@@ -2046,3 +2081,33 @@ func (storage *SStorage) GetDetailsHardwareInfo(ctx context.Context, userCred mc
func (storage *SStorage) PerformSetHardwareInfo(ctx context.Context, userCred mcclient.TokenCredential, _ jsonutils.JSONObject, data *api.StorageHardwareInfo) (*api.StorageHardwareInfo, error) {
return data, storage.setHardwareInfo(ctx, userCred, data)
}
func StoragesCleanRecycleDiskfiles(ctx context.Context, userCred mcclient.TokenCredential, isStart bool) {
// get shared storages
q := StorageManager.Query().IsNullOrEmpty("manager_id")
q = q.In("storage_type", api.SHARED_STORAGE)
storages := make([]SStorage, 0)
err := q.All(&storages)
if err != nil {
log.Errorf("StoragesCleanRecycleDiskfiles failed get storages %s", err)
return
}
for i := range storages {
storages[i].SetModelManager(StorageManager, &storages[i])
log.Infof("storage %s start clean recycle diskfiles", storages[i].GetName())
host, err := storages[i].GetMasterHost()
if err != nil {
log.Errorf("StoragesCleanRecycleDiskfiles storage %s failed get master host: %s", storages[i].GetName(), err)
continue
}
url := fmt.Sprintf("/storages/%s/clean-recycle-diskfiles", storages[i].Id)
body := jsonutils.NewDict()
_, err = host.Request(ctx, userCred, "POST", url, mcclient.GetTokenHeaders(userCred), body)
if err != nil {
log.Errorf("StoragesCleanRecycleDiskfiles storage %s request failed %s", storages[i].GetName(), err)
continue
}
}
}
pkg/compute/service/service.go
+3
View File
@@ -198,6 +198,9 @@ func StartServiceWithJobs(jobs func(cron *cronman.SCronJobManager)) {
cron.AddJobEveryFewHour("InspectAllTemplate", 1, 0, 0, models.GuestTemplateManager.InspectAllTemplate, true)
cron.AddJobEveryFewHour("CheckBillingResourceExpireAt", 1, 0, 0, models.CheckBillingResourceExpireAt, true)
cron.AddJobEveryFewDays(
"CleanRecycleDiskFiles", 1, 3, 0, 0, models.StoragesCleanRecycleDiskfiles, false)
if jobs != nil {
jobs(cron)
}
pkg/compute/storagedrivers/slvm.go
+3 -12
View File
@@ -47,18 +47,9 @@ func (s *SSLVMStorageDriver) ValidateCreateData(ctx context.Context, userCred mc
if len(input.SLVMVgName) == 0 {
return httperrors.NewMissingParameterError("slvm_vg_name")
}
if input.Lvmlockd {
input.MasterHost = ""
}
if !input.Lvmlockd && len(input.MasterHost) == 0 {
return httperrors.NewMissingParameterError("master_host")
}
if input.MasterHost != "" {
host, err := models.HostManager.FetchByIdOrName(ctx, userCred, input.MasterHost)
if err != nil {
return httperrors.NewInputParameterError("get host %s failed", input.MasterHost)
}
input.MasterHost = host.GetId()
if !input.Lvmlockd {
return httperrors.NewMissingParameterError("lvm_lockd")
}
storages := []models.SStorage{}
pkg/hostman/hostinfo/hostpinger/hostpinger.go
+11 -2
View File
@@ -39,7 +39,9 @@ type SHostPingTask struct {
running bool
host hostutils.IHost
lastStatAt time.Time
// masterHostStorages for shared storages
masterHostStorages []string
lastStatAt time.Time
}
type SEndpoint struct {
@@ -106,7 +108,7 @@ func (p *SHostPingTask) payload() api.SHostPingInput {
}
p.lastStatAt = now
data = storageman.GatherHostStorageStats()
data = storageman.GatherHostStorageStats(p.masterHostStorages)
data.WithData = true
info, err := mem.VirtualMemory()
if err != nil {
@@ -136,6 +138,13 @@ func (p *SHostPingTask) ping(div int, hostId string) error {
// if err != nil {
// Instance().setHostname(name)
// }
if res.Contains("master_host_storages") {
storages := make([]string, 0)
res.Unmarshal(&storages, "master_host_storages")
p.masterHostStorages = storages
}
catalog, err := res.Get("catalog")
if err == nil {
cl := make(mcclient.KeystoneServiceCatalogV3, 0)
pkg/hostman/storageman/core.go
+11 -1
View File
@@ -468,6 +468,9 @@ func CleanRecycleDiskfiles(ctx context.Context, userCred mcclient.TokenCredentia
return
}
for _, storage := range storageManager.Storages {
if utils.IsInStringArray(storage.StorageType(), api.SHARED_STORAGE) {
continue
}
storage.CleanRecycleDiskfiles(ctx)
}
}
@@ -491,11 +494,18 @@ func CleanImageCachefiles(ctx context.Context, userCred mcclient.TokenCredential
// }
}
func GatherHostStorageStats() api.SHostPingInput {
func GatherHostStorageStats(reportSharedStorages []string) api.SHostPingInput {
stats := api.SHostPingInput{}
stats.RootPartitionUsedCapacityMb = GetRootPartUsedCapacity()
manager := GetManager()
log.Debugf("report shared storages %s", reportSharedStorages)
for i := 0; i < len(manager.Storages); i++ {
if utils.IsInStringArray(manager.Storages[i].StorageType(), api.SHARED_STORAGE) &&
!utils.IsInStringArray(manager.Storages[i].GetId(), reportSharedStorages) {
log.Debugf("skip report storage %s", manager.Storages[i].GetId())
continue
}
iS := manager.Storages[i]
stat, err := iS.SyncStorageSize()
if err != nil {
pkg/hostman/storageman/storagehandler/storagehandler.go
+16 -1
View File
@@ -84,6 +84,9 @@ func AddStorageHandler(prefix string, app *appsrv.Application) {
app.AddHandler("POST",
fmt.Sprintf("%s/%s/sync-backup-storage", prefix, keyWords),
auth.Authenticate(storageSyncBackupStorage))
app.AddHandler("POST",
fmt.Sprintf("%s/%s/<storageId>/clean-recycle-diskfiles", prefix, keyWords),
auth.Authenticate(storageCleanRecycleDiskfiles))
}
}
@@ -494,7 +497,7 @@ func storageDeleteSnapshots(ctx context.Context, w http.ResponseWriter, r *http.
var storageId = params["<storageId>"]
storage := storageman.GetManager().GetStorage(storageId)
if storage == nil {
hostutils.Response(ctx, w, httperrors.NewNotFoundError("Stroage Not found"))
hostutils.Response(ctx, w, httperrors.NewNotFoundError("Storage Not found"))
return
}
diskId, err := body.GetString("disk_id")
@@ -517,3 +520,15 @@ func storageDeleteSnapshots(ctx context.Context, w http.ResponseWriter, r *http.
hostutils.DelayTask(ctx, storage.DeleteSnapshots, input)
hostutils.ResponseOk(ctx, w)
}
func storageCleanRecycleDiskfiles(ctx context.Context, w http.ResponseWriter, r *http.Request) {
params, _, _ := appsrv.FetchEnv(ctx, w, r)
var storageId = params["<storageId>"]
storage := storageman.GetManager().GetStorage(storageId)
if storage == nil {
hostutils.Response(ctx, w, httperrors.NewNotFoundError("Storage Not found"))
return
}
go storage.CleanRecycleDiskfiles(ctx)
hostutils.ResponseOk(ctx, w)
}