mirror of
https://github.com/yunionio/cloudpods.git
synced 2026-10-29 17:58:36 +08:00
fix: reduce sync workdload by not always sync VM disk/nics/eips & images
This commit is contained in:
@@ -3,6 +3,7 @@ package models
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"math/rand"
|
||||
"time"
|
||||
|
||||
"yunion.io/x/jsonutils"
|
||||
@@ -48,6 +49,8 @@ type SCloudproviderregion struct {
|
||||
|
||||
// SyncIntervalSeconds int `list:"admin"`
|
||||
SyncResults jsonutils.JSONObject `list:"admin"`
|
||||
|
||||
LastDeepSyncAt time.Time `list:"admin"`
|
||||
}
|
||||
|
||||
func (joint *SCloudproviderregion) Master() db.IStandaloneModel {
|
||||
@@ -108,14 +111,16 @@ func (self *SCloudproviderregion) getSyncIntervalSeconds(account *SCloudaccount)
|
||||
|
||||
func (self *SCloudproviderregion) getExtraDetails(extra *jsonutils.JSONDict) *jsonutils.JSONDict {
|
||||
account := self.GetAccount()
|
||||
extra.Add(jsonutils.NewString(account.Id), "cloudaccount_id")
|
||||
extra.Add(jsonutils.NewString(account.Name), "cloudaccount")
|
||||
if account.EnableAutoSync {
|
||||
extra.Add(jsonutils.JSONTrue, "enable_auto_sync")
|
||||
} else {
|
||||
extra.Add(jsonutils.JSONFalse, "enable_auto_sync")
|
||||
if account != nil {
|
||||
extra.Add(jsonutils.NewString(account.Id), "cloudaccount_id")
|
||||
extra.Add(jsonutils.NewString(account.Name), "cloudaccount")
|
||||
if account.EnableAutoSync {
|
||||
extra.Add(jsonutils.JSONTrue, "enable_auto_sync")
|
||||
} else {
|
||||
extra.Add(jsonutils.JSONFalse, "enable_auto_sync")
|
||||
}
|
||||
extra.Add(jsonutils.NewInt(int64(self.getSyncIntervalSeconds(account))), "sync_interval_seconds")
|
||||
}
|
||||
extra.Add(jsonutils.NewInt(int64(self.getSyncIntervalSeconds(account))), "sync_interval_seconds")
|
||||
return extra
|
||||
}
|
||||
|
||||
@@ -199,8 +204,8 @@ func (self *SCloudproviderregion) markSyncing(userCred mcclient.TokenCredential)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (self *SCloudproviderregion) markEndSync(ctx context.Context, userCred mcclient.TokenCredential, syncResults SSyncResultSet) error {
|
||||
err := self.markEndSyncInternal(userCred, syncResults)
|
||||
func (self *SCloudproviderregion) markEndSync(ctx context.Context, userCred mcclient.TokenCredential, syncResults SSyncResultSet, deepSync bool) error {
|
||||
err := self.markEndSyncInternal(userCred, syncResults, deepSync)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -211,11 +216,14 @@ func (self *SCloudproviderregion) markEndSync(ctx context.Context, userCred mccl
|
||||
return nil
|
||||
}
|
||||
|
||||
func (self *SCloudproviderregion) markEndSyncInternal(userCred mcclient.TokenCredential, syncResults SSyncResultSet) error {
|
||||
func (self *SCloudproviderregion) markEndSyncInternal(userCred mcclient.TokenCredential, syncResults SSyncResultSet, deepSync bool) error {
|
||||
_, err := db.Update(self, func() error {
|
||||
self.SyncStatus = CLOUD_PROVIDER_SYNC_STATUS_IDLE
|
||||
self.LastSyncEndAt = timeutils.UtcNow()
|
||||
self.SyncResults = jsonutils.Marshal(syncResults)
|
||||
if deepSync {
|
||||
self.LastDeepSyncAt = timeutils.UtcNow()
|
||||
}
|
||||
return nil
|
||||
})
|
||||
if err != nil {
|
||||
@@ -245,7 +253,7 @@ func (self *SCloudproviderregion) DoSync(ctx context.Context, userCred mcclient.
|
||||
syncResults := SSyncResultSet{}
|
||||
|
||||
self.markSyncing(userCred)
|
||||
defer self.markEndSync(ctx, userCred, syncResults)
|
||||
defer self.markEndSync(ctx, userCred, syncResults, syncRange.DeepSync)
|
||||
|
||||
localRegion := self.GetRegion()
|
||||
provider := self.GetProvider()
|
||||
@@ -255,6 +263,13 @@ func (self *SCloudproviderregion) DoSync(ctx context.Context, userCred mcclient.
|
||||
return err
|
||||
}
|
||||
|
||||
if !syncRange.DeepSync {
|
||||
intval := self.getSyncIntervalSeconds(nil)
|
||||
if self.LastDeepSyncAt.IsZero() || (time.Now().Sub(self.LastDeepSyncAt) > time.Duration(intval)*time.Second*8 && rand.Float32() < 0.5) {
|
||||
syncRange.DeepSync = true
|
||||
}
|
||||
}
|
||||
|
||||
if localRegion.isManaged() {
|
||||
remoteRegion, err := driver.GetIRegionById(localRegion.ExternalId)
|
||||
if err == nil {
|
||||
@@ -268,7 +283,7 @@ func (self *SCloudproviderregion) DoSync(ctx context.Context, userCred mcclient.
|
||||
log.Errorf("dosync fail %s", err)
|
||||
}
|
||||
|
||||
log.Debugf("%s", jsonutils.Marshal(syncResults))
|
||||
log.Debugf("dosync result: %s", jsonutils.Marshal(syncResults))
|
||||
|
||||
return err
|
||||
}
|
||||
@@ -312,7 +327,8 @@ func (cpr *SCloudproviderregion) needAutoSync() bool {
|
||||
region := cpr.GetRegion()
|
||||
log.Debugf("empty region %s! no need to check so frequently", region.GetName())
|
||||
}
|
||||
if time.Now().Sub(cpr.LastSyncEndAt) > time.Duration(intval)*time.Second {
|
||||
if time.Now().Sub(cpr.LastSyncEndAt) > time.Duration(intval)*time.Second && rand.Float32() < 0.6 {
|
||||
// add randomness
|
||||
return true
|
||||
}
|
||||
return false
|
||||
|
||||
@@ -326,6 +326,7 @@ func (self *SCloudprovider) saveProject(userCred mcclient.TokenCredential, proje
|
||||
type SSyncRange struct {
|
||||
Force bool
|
||||
FullSync bool
|
||||
DeepSync bool
|
||||
// ProjectSync bool
|
||||
|
||||
Region []string
|
||||
@@ -513,7 +514,7 @@ func (self *SCloudprovider) PerformChangeProject(ctx context.Context, userCred m
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
return nil, self.StartSyncCloudProviderInfoTask(ctx, userCred, &SSyncRange{FullSync: true}, "")
|
||||
return nil, self.StartSyncCloudProviderInfoTask(ctx, userCred, &SSyncRange{FullSync: true, DeepSync: true}, "")
|
||||
}
|
||||
|
||||
func (self *SCloudprovider) markStartSync(userCred mcclient.TokenCredential) error {
|
||||
|
||||
@@ -463,7 +463,7 @@ func syncHostVMs(ctx context.Context, userCred mcclient.TokenCredential, syncRes
|
||||
log.Errorf(msg)
|
||||
return
|
||||
}
|
||||
localVMs, remoteVMs, result := localHost.SyncHostVMs(ctx, userCred, driver, vms, provider.ProjectId)
|
||||
syncVMPairs, result := localHost.SyncHostVMs(ctx, userCred, driver, vms, provider.ProjectId)
|
||||
|
||||
syncResults.Add(GuestManager, result)
|
||||
|
||||
@@ -473,17 +473,21 @@ func syncHostVMs(ctx context.Context, userCred mcclient.TokenCredential, syncRes
|
||||
if result.IsError() {
|
||||
return
|
||||
}
|
||||
|
||||
// db.OpsLog.LogEvent(provider, db.ACT_SYNC_HOST_COMPLETE, msg, userCred)
|
||||
// logclient.AddActionLog(provider, getAction(task.Params), notes, task.UserCred, true)
|
||||
for i := 0; i < len(localVMs); i += 1 {
|
||||
for i := 0; i < len(syncVMPairs); i += 1 {
|
||||
if !syncVMPairs[i].IsNew && !syncRange.DeepSync {
|
||||
continue
|
||||
}
|
||||
func() {
|
||||
lockman.LockObject(ctx, &localVMs[i])
|
||||
defer lockman.ReleaseObject(ctx, &localVMs[i])
|
||||
lockman.LockObject(ctx, syncVMPairs[i].Local)
|
||||
defer lockman.ReleaseObject(ctx, syncVMPairs[i].Local)
|
||||
|
||||
syncVMNics(ctx, userCred, provider, localHost, &localVMs[i], remoteVMs[i])
|
||||
syncVMDisks(ctx, userCred, provider, driver, localHost, &localVMs[i], remoteVMs[i], syncRange)
|
||||
syncVMEip(ctx, userCred, provider, &localVMs[i], remoteVMs[i])
|
||||
syncVMSecgroups(ctx, userCred, provider, &localVMs[i], remoteVMs[i])
|
||||
syncVMNics(ctx, userCred, provider, localHost, syncVMPairs[i].Local, syncVMPairs[i].Remote)
|
||||
syncVMDisks(ctx, userCred, provider, driver, localHost, syncVMPairs[i].Local, syncVMPairs[i].Remote, syncRange)
|
||||
syncVMEip(ctx, userCred, provider, syncVMPairs[i].Local, syncVMPairs[i].Remote)
|
||||
syncVMSecgroups(ctx, userCred, provider, syncVMPairs[i].Local, syncVMPairs[i].Remote)
|
||||
|
||||
}()
|
||||
}
|
||||
@@ -806,14 +810,16 @@ func syncPublicCloudProviderInfo(
|
||||
syncRegionLoadbalancerCertificates(ctx, userCred, syncResults, provider, localRegion, remoteRegion, syncRange)
|
||||
syncRegionLoadbalancers(ctx, userCred, syncResults, provider, localRegion, remoteRegion, syncRange)
|
||||
|
||||
log.Debugf("storageCachePairs count %d", len(storageCachePairs))
|
||||
for i := range storageCachePairs {
|
||||
result := storageCachePairs[i].syncCloudImages(ctx, userCred)
|
||||
if syncRange.DeepSync {
|
||||
log.Debugf("storageCachePairs count %d", len(storageCachePairs))
|
||||
for i := range storageCachePairs {
|
||||
result := storageCachePairs[i].syncCloudImages(ctx, userCred)
|
||||
|
||||
syncResults.Add(StoragecachedimageManager, result)
|
||||
syncResults.Add(StoragecachedimageManager, result)
|
||||
|
||||
msg := result.Result()
|
||||
log.Infof("syncCloudImages result: %s", msg)
|
||||
msg := result.Result()
|
||||
log.Infof("syncCloudImages result: %s", msg)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
@@ -870,12 +876,14 @@ func syncOnPremiseCloudProviderInfo(
|
||||
syncHostVMs(ctx, userCred, syncResults, provider, driver, &localHosts[i], remoteHosts[i], syncRange)
|
||||
}
|
||||
|
||||
log.Debugf("storageCachePairs count %d", len(storageCachePairs))
|
||||
for i := range storageCachePairs {
|
||||
result := storageCachePairs[i].syncCloudImages(ctx, userCred)
|
||||
syncResults.Add(StoragecachedimageManager, result)
|
||||
msg := result.Result()
|
||||
log.Infof("syncCloudImages result: %s", msg)
|
||||
if syncRange.DeepSync {
|
||||
log.Debugf("storageCachePairs count %d", len(storageCachePairs))
|
||||
for i := range storageCachePairs {
|
||||
result := storageCachePairs[i].syncCloudImages(ctx, userCred)
|
||||
syncResults.Add(StoragecachedimageManager, result)
|
||||
msg := result.Result()
|
||||
log.Infof("syncCloudImages result: %s", msg)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
+23
-10
@@ -1729,14 +1729,19 @@ func (self *SHost) newCloudHostWire(ctx context.Context, userCred mcclient.Token
|
||||
return err
|
||||
}
|
||||
|
||||
func (self *SHost) SyncHostVMs(ctx context.Context, userCred mcclient.TokenCredential, iprovider cloudprovider.ICloudProvider, vms []cloudprovider.ICloudVM, projectId string) ([]SGuest, []cloudprovider.ICloudVM, compare.SyncResult) {
|
||||
type SGuestSyncResult struct {
|
||||
Local *SGuest
|
||||
Remote cloudprovider.ICloudVM
|
||||
IsNew bool
|
||||
}
|
||||
|
||||
func (self *SHost) SyncHostVMs(ctx context.Context, userCred mcclient.TokenCredential, iprovider cloudprovider.ICloudProvider, vms []cloudprovider.ICloudVM, projectId string) ([]SGuestSyncResult, compare.SyncResult) {
|
||||
syncOwnerId := projectId
|
||||
|
||||
lockman.LockClass(ctx, GuestManager, syncOwnerId)
|
||||
defer lockman.ReleaseClass(ctx, GuestManager, syncOwnerId)
|
||||
|
||||
localVMs := make([]SGuest, 0)
|
||||
remoteVMs := make([]cloudprovider.ICloudVM, 0)
|
||||
syncVMPairs := make([]SGuestSyncResult, 0)
|
||||
syncResult := compare.SyncResult{}
|
||||
|
||||
dbVMs := self.GetGuests()
|
||||
@@ -1744,7 +1749,7 @@ func (self *SHost) SyncHostVMs(ctx context.Context, userCred mcclient.TokenCrede
|
||||
for i := range dbVMs {
|
||||
if taskman.TaskManager.IsInTask(&dbVMs[i]) {
|
||||
syncResult.Error(fmt.Errorf("object in task"))
|
||||
return nil, nil, syncResult
|
||||
return nil, syncResult
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1756,7 +1761,7 @@ func (self *SHost) SyncHostVMs(ctx context.Context, userCred mcclient.TokenCrede
|
||||
err := compare.CompareSets(dbVMs, vms, &removed, &commondb, &commonext, &added)
|
||||
if err != nil {
|
||||
syncResult.Error(err)
|
||||
return nil, nil, syncResult
|
||||
return nil, syncResult
|
||||
}
|
||||
|
||||
for i := 0; i < len(removed); i += 1 {
|
||||
@@ -1774,8 +1779,12 @@ func (self *SHost) SyncHostVMs(ctx context.Context, userCred mcclient.TokenCrede
|
||||
syncResult.UpdateError(err)
|
||||
} else {
|
||||
syncMetadata(ctx, userCred, &commondb[i], commonext[i])
|
||||
localVMs = append(localVMs, commondb[i])
|
||||
remoteVMs = append(remoteVMs, commonext[i])
|
||||
syncVMPair := SGuestSyncResult{
|
||||
Local: &commondb[i],
|
||||
Remote: commonext[i],
|
||||
IsNew: false,
|
||||
}
|
||||
syncVMPairs = append(syncVMPairs, syncVMPair)
|
||||
syncResult.Update()
|
||||
}
|
||||
}
|
||||
@@ -1797,13 +1806,17 @@ func (self *SHost) SyncHostVMs(ctx context.Context, userCred mcclient.TokenCrede
|
||||
syncResult.AddError(err)
|
||||
} else {
|
||||
syncMetadata(ctx, userCred, new, added[i])
|
||||
localVMs = append(localVMs, *new)
|
||||
remoteVMs = append(remoteVMs, added[i])
|
||||
syncVMPair := SGuestSyncResult{
|
||||
Local: new,
|
||||
Remote: added[i],
|
||||
IsNew: true,
|
||||
}
|
||||
syncVMPairs = append(syncVMPairs, syncVMPair)
|
||||
syncResult.Add()
|
||||
}
|
||||
}
|
||||
|
||||
return localVMs, remoteVMs, syncResult
|
||||
return syncVMPairs, syncResult
|
||||
}
|
||||
|
||||
func (self *SHost) getNetworkOfIPOnHost(ipAddr string) (*SNetwork, error) {
|
||||
|
||||
@@ -42,6 +42,7 @@ func (self *CloudAccountSyncInfoTask) OnInit(ctx context.Context, obj db.IStanda
|
||||
syncRangeJson.Unmarshal(&syncRange)
|
||||
} else {
|
||||
syncRange.FullSync = true
|
||||
syncRange.DeepSync = true
|
||||
}
|
||||
|
||||
if !syncRange.NeedSyncInfo() {
|
||||
|
||||
@@ -12,6 +12,7 @@ func init() {
|
||||
"Cloudregion_ID", "CloudRegion",
|
||||
"Enabled", "Sync_Status",
|
||||
"Last_Sync", "Last_Sync_End_At", "Auto_Sync",
|
||||
"last_deep_sync_at",
|
||||
"Sync_Results"},
|
||||
[]string{},
|
||||
&Cloudproviders,
|
||||
|
||||
Reference in New Issue
Block a user