cloudsync.go 116 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737173817391740174117421743174417451746174717481749175017511752175317541755175617571758175917601761176217631764176517661767176817691770177117721773177417751776177717781779178017811782178317841785178617871788178917901791179217931794179517961797179817991800180118021803180418051806180718081809181018111812181318141815181618171818181918201821182218231824182518261827182818291830183118321833183418351836183718381839184018411842184318441845184618471848184918501851185218531854185518561857185818591860186118621863186418651866186718681869187018711872187318741875187618771878187918801881188218831884188518861887188818891890189118921893189418951896189718981899190019011902190319041905190619071908190919101911191219131914191519161917191819191920192119221923192419251926192719281929193019311932193319341935193619371938193919401941194219431944194519461947194819491950195119521953195419551956195719581959196019611962196319641965196619671968196919701971197219731974197519761977197819791980198119821983198419851986198719881989199019911992199319941995199619971998199920002001200220032004200520062007200820092010201120122013201420152016201720182019202020212022202320242025202620272028202920302031203220332034203520362037203820392040204120422043204420452046204720482049205020512052205320542055205620572058205920602061206220632064206520662067206820692070207120722073207420752076207720782079208020812082208320842085208620872088208920902091209220932094209520962097209820992100210121022103210421052106210721082109211021112112211321142115211621172118211921202121212221232124212521262127212821292130213121322133213421352136213721382139214021412142214321442145214621472148214921502151215221532154215521562157215821592160216121622163216421652166216721682169217021712172217321742175217621772178217921802181218221832184218521862187218821892190219121922193219421952196219721982199220022012202220322042205220622072208220922102211221222132214221522162217221822192220222122222223222422252226222722282229223022312232223322342235223622372238223922402241224222432244224522462247224822492250225122522253225422552256225722582259226022612262226322642265226622672268226922702271227222732274227522762277227822792280228122822283228422852286228722882289229022912292229322942295229622972298229923002301230223032304230523062307230823092310231123122313231423152316231723182319232023212322232323242325232623272328232923302331233223332334233523362337233823392340234123422343234423452346234723482349235023512352235323542355235623572358235923602361236223632364236523662367236823692370237123722373237423752376237723782379238023812382238323842385238623872388238923902391239223932394239523962397239823992400240124022403240424052406240724082409241024112412241324142415241624172418241924202421242224232424242524262427242824292430243124322433243424352436243724382439244024412442244324442445244624472448244924502451245224532454245524562457245824592460246124622463246424652466246724682469247024712472247324742475247624772478247924802481248224832484248524862487248824892490249124922493249424952496249724982499250025012502250325042505250625072508250925102511251225132514251525162517251825192520252125222523252425252526252725282529253025312532253325342535253625372538253925402541254225432544254525462547254825492550255125522553255425552556255725582559256025612562256325642565256625672568256925702571257225732574257525762577257825792580258125822583258425852586258725882589259025912592259325942595259625972598259926002601260226032604260526062607260826092610261126122613261426152616261726182619262026212622262326242625262626272628262926302631263226332634263526362637263826392640264126422643264426452646264726482649265026512652265326542655265626572658265926602661266226632664266526662667266826692670267126722673267426752676267726782679268026812682268326842685268626872688268926902691269226932694269526962697269826992700270127022703270427052706270727082709271027112712271327142715271627172718271927202721272227232724272527262727272827292730273127322733273427352736273727382739274027412742274327442745274627472748274927502751275227532754275527562757275827592760276127622763276427652766276727682769277027712772277327742775277627772778277927802781278227832784278527862787278827892790279127922793279427952796279727982799280028012802280328042805280628072808280928102811281228132814281528162817281828192820282128222823282428252826282728282829283028312832283328342835283628372838283928402841284228432844284528462847284828492850285128522853285428552856285728582859286028612862286328642865286628672868286928702871287228732874287528762877287828792880288128822883288428852886288728882889289028912892289328942895289628972898289929002901290229032904290529062907290829092910291129122913291429152916291729182919292029212922292329242925292629272928292929302931293229332934293529362937293829392940294129422943294429452946294729482949295029512952295329542955295629572958295929602961296229632964296529662967296829692970297129722973297429752976297729782979298029812982298329842985298629872988298929902991299229932994299529962997299829993000300130023003300430053006300730083009301030113012301330143015301630173018301930203021302230233024302530263027302830293030303130323033303430353036303730383039304030413042304330443045304630473048304930503051305230533054305530563057305830593060306130623063306430653066306730683069307030713072307330743075307630773078307930803081308230833084308530863087308830893090309130923093309430953096309730983099310031013102310331043105310631073108310931103111311231133114311531163117311831193120312131223123312431253126312731283129313031313132313331343135313631373138313931403141314231433144314531463147314831493150315131523153315431553156315731583159316031613162316331643165316631673168316931703171317231733174317531763177317831793180318131823183318431853186318731883189319031913192319331943195319631973198319932003201320232033204320532063207320832093210321132123213321432153216321732183219322032213222322332243225322632273228322932303231323232333234323532363237323832393240324132423243324432453246324732483249325032513252325332543255325632573258325932603261326232633264326532663267326832693270
  1. // Copyright 2019 Yunion
  2. //
  3. // Licensed under the Apache License, Version 2.0 (the "License");
  4. // you may not use this file except in compliance with the License.
  5. // You may obtain a copy of the License at
  6. //
  7. // http://www.apache.org/licenses/LICENSE-2.0
  8. //
  9. // Unless required by applicable law or agreed to in writing, software
  10. // distributed under the License is distributed on an "AS IS" BASIS,
  11. // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12. // See the License for the specific language governing permissions and
  13. // limitations under the License.
  14. package models
  15. import (
  16. "context"
  17. "fmt"
  18. "time"
  19. "yunion.io/x/cloudmux/pkg/cloudprovider"
  20. "yunion.io/x/log"
  21. "yunion.io/x/pkg/errors"
  22. "yunion.io/x/pkg/util/compare"
  23. "yunion.io/x/pkg/utils"
  24. "yunion.io/x/sqlchemy"
  25. "yunion.io/x/onecloud/pkg/apis"
  26. api "yunion.io/x/onecloud/pkg/apis/compute"
  27. "yunion.io/x/onecloud/pkg/cloudcommon/db"
  28. "yunion.io/x/onecloud/pkg/cloudcommon/db/lockman"
  29. "yunion.io/x/onecloud/pkg/mcclient"
  30. "yunion.io/x/onecloud/pkg/mcclient/auth"
  31. "yunion.io/x/onecloud/pkg/mcclient/modules/scheduler"
  32. "yunion.io/x/onecloud/pkg/util/logclient"
  33. )
  34. type SSyncableBaseResource struct {
  35. SyncStatus string `width:"10" charset:"ascii" default:"idle" list:"domain"`
  36. LastSync time.Time `list:"domain"` // = Column(DateTime, nullable=True)
  37. LastSyncEndAt time.Time `list:"domain"`
  38. }
  39. func (self SSyncableBaseResource) GetLastSyncCost() string {
  40. if !self.LastSync.IsZero() && !self.LastSyncEndAt.IsZero() {
  41. return self.LastSyncEndAt.Sub(self.LastSync).Round(time.Second).String()
  42. }
  43. return ""
  44. }
  45. type SSyncableBaseResourceManager struct{}
  46. func (self *SSyncableBaseResource) CanSync() bool {
  47. if self.SyncStatus == api.CLOUD_PROVIDER_SYNC_STATUS_QUEUED || self.SyncStatus == api.CLOUD_PROVIDER_SYNC_STATUS_SYNCING {
  48. if self.LastSync.IsZero() || time.Now().Sub(self.LastSync) > time.Minute*30 {
  49. return true
  50. }
  51. return false
  52. }
  53. return true
  54. }
  55. func (manager *SSyncableBaseResourceManager) ListItemFilter(
  56. ctx context.Context,
  57. q *sqlchemy.SQuery,
  58. userCred mcclient.TokenCredential,
  59. query api.SyncableBaseResourceListInput,
  60. ) (*sqlchemy.SQuery, error) {
  61. if len(query.SyncStatus) > 0 {
  62. q = q.In("sync_status", query.SyncStatus)
  63. }
  64. return q, nil
  65. }
  66. type sStoragecacheSyncPair struct {
  67. local *SStoragecache
  68. region *SCloudregion
  69. remote cloudprovider.ICloudStoragecache
  70. isNew bool
  71. }
  72. func (self *sStoragecacheSyncPair) isValid() bool {
  73. return self.local != nil && self.region != nil && self.remote != nil
  74. }
  75. func (pair *sStoragecacheSyncPair) syncCloudImages(ctx context.Context, userCred mcclient.TokenCredential, xor bool) compare.SyncResult {
  76. return pair.local.SyncCloudImages(ctx, userCred, pair.remote, pair.region, xor)
  77. }
  78. func isInCache(pairs []sStoragecacheSyncPair, localCacheId string) bool {
  79. // log.Debugf("isInCache %d %s", len(pairs), localCacheId)
  80. for i := range pairs {
  81. if pairs[i].local.Id == localCacheId {
  82. return true
  83. }
  84. }
  85. return false
  86. }
  87. func syncRegionQuotas(ctx context.Context, userCred mcclient.TokenCredential, syncResults SSyncResultSet, driver cloudprovider.ICloudProvider, provider *SCloudprovider, localRegion *SCloudregion, remoteRegion cloudprovider.ICloudRegion, xor bool) error {
  88. quotas, err := func() ([]cloudprovider.ICloudQuota, error) {
  89. defer syncResults.AddRequestCost(CloudproviderQuotaManager)()
  90. return remoteRegion.GetICloudQuotas()
  91. }()
  92. if err != nil {
  93. if errors.Cause(err) == cloudprovider.ErrNotImplemented || errors.Cause(err) == cloudprovider.ErrNotSupported {
  94. return nil
  95. }
  96. msg := fmt.Sprintf("GetICloudQuotas for region %s provider %s failed %s", remoteRegion.GetName(), provider.Name, err)
  97. log.Errorf("%s", msg)
  98. return err
  99. }
  100. result := func() compare.SyncResult {
  101. defer syncResults.AddSqlCost(CloudproviderQuotaManager)()
  102. return CloudproviderQuotaManager.SyncQuotas(ctx, userCred, provider.GetOwnerId(), provider, localRegion, api.CLOUD_PROVIDER_QUOTA_RANGE_CLOUDREGION, quotas, xor)
  103. }()
  104. syncResults.Add(CloudproviderQuotaManager, result)
  105. msg := result.Result()
  106. notes := fmt.Sprintf("SyncQuotas for region %s provider %s result: %s", localRegion.Name, provider.Name, msg)
  107. provider.SyncError(result, notes, userCred)
  108. log.Infof("%s", notes)
  109. if result.IsError() {
  110. return fmt.Errorf("%s", msg)
  111. }
  112. return nil
  113. }
  114. func syncRegionZones(ctx context.Context, userCred mcclient.TokenCredential, syncResults SSyncResultSet, provider *SCloudprovider, localRegion *SCloudregion, remoteRegion cloudprovider.ICloudRegion, xor bool) ([]SZone, []cloudprovider.ICloudZone, error) {
  115. zones, err := func() ([]cloudprovider.ICloudZone, error) {
  116. defer syncResults.AddRequestCost(ZoneManager)()
  117. return remoteRegion.GetIZones()
  118. }()
  119. if err != nil {
  120. msg := fmt.Sprintf("GetZones for region %s provider %s failed %s", remoteRegion.GetName(), provider.Name, err)
  121. log.Errorf("%s", msg)
  122. return nil, nil, err
  123. }
  124. localZones, remoteZones, result := func() ([]SZone, []cloudprovider.ICloudZone, compare.SyncResult) {
  125. defer syncResults.AddSqlCost(ZoneManager)()
  126. return ZoneManager.SyncZones(ctx, userCred, localRegion, zones, provider, xor)
  127. }()
  128. syncResults.Add(ZoneManager, result)
  129. msg := result.Result()
  130. notes := fmt.Sprintf("SyncZones for region %s provider %s result: %s", localRegion.Name, provider.Name, msg)
  131. log.Infof("%s", notes)
  132. provider.SyncError(result, notes, userCred)
  133. db.OpsLog.LogEvent(provider, db.ACT_SYNC_HOST_COMPLETE, msg, userCred)
  134. return localZones, remoteZones, nil
  135. }
  136. func syncRegionSkus(ctx context.Context, userCred mcclient.TokenCredential, localRegion *SCloudregion, xor bool) {
  137. if localRegion == nil {
  138. log.Debugf("local region is nil, skipp...")
  139. return
  140. }
  141. regionId := localRegion.GetId()
  142. if len(regionId) == 0 {
  143. log.Debugf("local region Id is empty, skip...")
  144. return
  145. }
  146. cnt, err := ServerSkuManager.GetSkuCountByRegion(regionId)
  147. if err != nil {
  148. log.Errorf("GetSkuCountByRegion fail %s", err)
  149. return
  150. }
  151. if cnt == 0 {
  152. // 提前同步instance type.如果同步失败可能导致vm 内存显示为0
  153. localRegion.StartSyncSkusTask(ctx, userCred, ServerSkuManager.Keyword())
  154. }
  155. if localRegion.GetDriver().IsSupportedElasticcache() {
  156. cnt, err = ElasticcacheSkuManager.GetSkuCountByRegion(regionId)
  157. if err != nil {
  158. log.Errorf("ElasticcacheSkuManager.GetSkuCountByRegion fail %s", err)
  159. return
  160. }
  161. if cnt == 0 {
  162. localRegion.StartSyncSkusTask(ctx, userCred, ElasticcacheSkuManager.Keyword())
  163. }
  164. }
  165. if localRegion.GetDriver().IsSupportedDBInstance() {
  166. cnt, err = DBInstanceSkuManager.GetSkuCountByRegion(regionId)
  167. if err != nil {
  168. log.Errorf("DBInstanceSkuManager.GetSkuCountByRegion fail %s", err)
  169. return
  170. }
  171. if cnt == 0 {
  172. localRegion.StartSyncSkusTask(ctx, userCred, DBInstanceSkuManager.Keyword())
  173. }
  174. }
  175. }
  176. func syncRegionEips(
  177. ctx context.Context,
  178. userCred mcclient.TokenCredential,
  179. syncResults SSyncResultSet,
  180. provider *SCloudprovider,
  181. localRegion *SCloudregion,
  182. remoteRegion cloudprovider.ICloudRegion,
  183. syncRange *SSyncRange,
  184. ) {
  185. eips, err := func() ([]cloudprovider.ICloudEIP, error) {
  186. defer syncResults.AddRequestCost(ElasticipManager)()
  187. return remoteRegion.GetIEips()
  188. }()
  189. if err != nil {
  190. if errors.Cause(err) == cloudprovider.ErrNotImplemented || errors.Cause(err) == cloudprovider.ErrNotSupported {
  191. return
  192. }
  193. msg := fmt.Sprintf("GetIEips for region %s provider %s failed %s", remoteRegion.GetName(), provider.Name, err)
  194. log.Errorf("%s", msg)
  195. return
  196. }
  197. result := func() compare.SyncResult {
  198. defer syncResults.AddSqlCost(ElasticipManager)()
  199. return ElasticipManager.SyncEips(ctx, userCred, provider, localRegion, eips, provider.GetOwnerId(), syncRange.Xor)
  200. }()
  201. syncResults.Add(ElasticipManager, result)
  202. msg := result.Result()
  203. notes := fmt.Sprintf("SyncEips for region %s provider %s result: %s", localRegion.Name, provider.Name, msg)
  204. log.Infof("%s", msg)
  205. provider.SyncError(result, notes, userCred)
  206. if result.IsError() {
  207. return
  208. }
  209. }
  210. func syncRegionBuckets(ctx context.Context, userCred mcclient.TokenCredential, syncResults SSyncResultSet, provider *SCloudprovider, localRegion *SCloudregion, remoteRegion cloudprovider.ICloudRegion, xor bool) {
  211. buckets, err := func() ([]cloudprovider.ICloudBucket, error) {
  212. defer syncResults.AddRequestCost(BucketManager)()
  213. return remoteRegion.GetIBuckets()
  214. }()
  215. if err != nil {
  216. if errors.Cause(err) == cloudprovider.ErrNotImplemented || errors.Cause(err) == cloudprovider.ErrNotSupported {
  217. return
  218. }
  219. msg := fmt.Sprintf("GetIBuckets for region %s provider %s failed %s", remoteRegion.GetName(), provider.Name, err)
  220. log.Errorf("%s", msg)
  221. return
  222. }
  223. result := func() compare.SyncResult {
  224. defer syncResults.AddSqlCost(BucketManager)()
  225. return BucketManager.syncBuckets(ctx, userCred, provider, localRegion, buckets, xor)
  226. }()
  227. syncResults.Add(BucketManager, result)
  228. msg := result.Result()
  229. notes := fmt.Sprintf("GetIBuckets for region %s provider %s result: %s", localRegion.Name, provider.Name, msg)
  230. log.Infof("%s", notes)
  231. provider.SyncError(result, notes, userCred)
  232. if result.IsError() {
  233. return
  234. }
  235. db.OpsLog.LogEvent(provider, db.ACT_SYNC_HOST_COMPLETE, msg, userCred)
  236. }
  237. func syncRegionVPCs(
  238. ctx context.Context,
  239. userCred mcclient.TokenCredential,
  240. syncResults SSyncResultSet,
  241. provider *SCloudprovider,
  242. localRegion *SCloudregion,
  243. remoteRegion cloudprovider.ICloudRegion,
  244. syncRange *SSyncRange,
  245. ) {
  246. vpcs, err := func() ([]cloudprovider.ICloudVpc, error) {
  247. defer syncResults.AddRequestCost(VpcManager)()
  248. return remoteRegion.GetIVpcs()
  249. }()
  250. if err != nil {
  251. if errors.Cause(err) == cloudprovider.ErrNotImplemented || errors.Cause(err) == cloudprovider.ErrNotSupported {
  252. return
  253. }
  254. msg := fmt.Sprintf("GetVpcs for region %s provider %s failed %s", remoteRegion.GetName(), provider.Name, err)
  255. log.Errorf("%s", msg)
  256. return
  257. }
  258. localVpcs, remoteVpcs, result := func() ([]SVpc, []cloudprovider.ICloudVpc, compare.SyncResult) {
  259. defer syncResults.AddSqlCost(VpcManager)()
  260. return VpcManager.SyncVPCs(ctx, userCred, provider, localRegion, vpcs, syncRange.Xor)
  261. }()
  262. syncResults.Add(VpcManager, result)
  263. msg := result.Result()
  264. notes := fmt.Sprintf("SyncVPCs for region %s provider %s result: %s", localRegion.Name, provider.Name, msg)
  265. log.Infof("%s", notes)
  266. provider.SyncError(result, notes, userCred)
  267. if result.IsError() {
  268. return
  269. }
  270. db.OpsLog.LogEvent(provider, db.ACT_SYNC_HOST_COMPLETE, msg, userCred)
  271. for j := 0; j < len(localVpcs); j += 1 {
  272. func() {
  273. // lock vpc
  274. lockman.LockObject(ctx, &localVpcs[j])
  275. defer lockman.ReleaseObject(ctx, &localVpcs[j])
  276. if localVpcs[j].Deleted {
  277. return
  278. }
  279. syncVpcWires(ctx, userCred, syncResults, provider, &localVpcs[j], remoteVpcs[j], nil, syncRange)
  280. if syncRange.IsNotSkipSyncResource(SecurityGroupManager) {
  281. syncVpcSecGroup(ctx, userCred, syncResults, provider, localRegion, &localVpcs[j], remoteVpcs[j], syncRange)
  282. }
  283. if syncRange.IsNotSkipSyncResource(NatGatewayManager) {
  284. syncVpcNatgateways(ctx, userCred, syncResults, provider, &localVpcs[j], remoteVpcs[j], syncRange)
  285. }
  286. syncVpcPeerConnections(ctx, userCred, syncResults, provider, &localVpcs[j], remoteVpcs[j], syncRange)
  287. syncVpcRouteTables(ctx, userCred, syncResults, provider, &localVpcs[j], remoteVpcs[j], syncRange)
  288. syncIPv6Gateways(ctx, userCred, syncResults, provider, &localVpcs[j], remoteVpcs[j], syncRange)
  289. }()
  290. }
  291. }
  292. func syncRegionAccessGroups(ctx context.Context, userCred mcclient.TokenCredential, syncResults SSyncResultSet, provider *SCloudprovider, localRegion *SCloudregion, remoteRegion cloudprovider.ICloudRegion, syncRange *SSyncRange) {
  293. accessGroups, err := func() ([]cloudprovider.ICloudAccessGroup, error) {
  294. defer syncResults.AddRequestCost(AccessGroupManager)()
  295. return remoteRegion.GetICloudAccessGroups()
  296. }()
  297. if err != nil {
  298. if errors.Cause(err) == cloudprovider.ErrNotImplemented || errors.Cause(err) == cloudprovider.ErrNotSupported {
  299. return
  300. }
  301. log.Errorf("GetICloudFileSystems for region %s provider %s error: %v", localRegion.Name, provider.Name, err)
  302. return
  303. }
  304. result := func() compare.SyncResult {
  305. defer syncResults.AddSqlCost(AccessGroupManager)()
  306. return localRegion.SyncAccessGroups(ctx, userCred, provider, accessGroups, syncRange.Xor)
  307. }()
  308. syncResults.Add(AccessGroupManager, result)
  309. msg := result.Result()
  310. notes := fmt.Sprintf("Sync Access Group for region %s provider %s result: %s", localRegion.Name, provider.Name, msg)
  311. log.Infof("%s", notes)
  312. provider.SyncError(result, notes, userCred)
  313. }
  314. func syncRegionFileSystems(
  315. ctx context.Context,
  316. userCred mcclient.TokenCredential,
  317. syncResults SSyncResultSet,
  318. provider *SCloudprovider,
  319. localRegion *SCloudregion,
  320. remoteRegion cloudprovider.ICloudRegion,
  321. syncRange *SSyncRange,
  322. ) {
  323. filesystems, err := func() ([]cloudprovider.ICloudFileSystem, error) {
  324. defer syncResults.AddRequestCost(FileSystemManager)()
  325. return remoteRegion.GetICloudFileSystems()
  326. }()
  327. if err != nil {
  328. if errors.Cause(err) == cloudprovider.ErrNotImplemented || errors.Cause(err) == cloudprovider.ErrNotSupported {
  329. return
  330. }
  331. log.Errorf("GetICloudFileSystems for region %s provider %s error: %v", localRegion.Name, provider.Name, err)
  332. return
  333. }
  334. localFSs, removeFSs, result := func() ([]SFileSystem, []cloudprovider.ICloudFileSystem, compare.SyncResult) {
  335. defer syncResults.AddSqlCost(FileSystemManager)()
  336. return localRegion.SyncFileSystems(ctx, userCred, provider, filesystems, syncRange.Xor)
  337. }()
  338. syncResults.Add(FileSystemManager, result)
  339. msg := result.Result()
  340. notes := fmt.Sprintf("Sync FileSystem for region %s provider %s result: %s", localRegion.Name, provider.Name, msg)
  341. log.Infof("%s", notes)
  342. provider.SyncError(result, notes, userCred)
  343. for j := 0; j < len(localFSs); j += 1 {
  344. func() {
  345. // lock file system
  346. lockman.LockObject(ctx, &localFSs[j])
  347. defer lockman.ReleaseObject(ctx, &localFSs[j])
  348. if localFSs[j].Deleted {
  349. return
  350. }
  351. syncFileSystemMountTargets(ctx, userCred, &localFSs[j], removeFSs[j], syncRange.Xor)
  352. }()
  353. }
  354. }
  355. func syncFileSystemMountTargets(
  356. ctx context.Context,
  357. userCred mcclient.TokenCredential,
  358. localFs *SFileSystem,
  359. remoteFs cloudprovider.ICloudFileSystem,
  360. xor bool,
  361. ) {
  362. mountTargets, err := remoteFs.GetMountTargets()
  363. if err != nil {
  364. if errors.Cause(err) == cloudprovider.ErrNotImplemented || errors.Cause(err) == cloudprovider.ErrNotSupported {
  365. return
  366. }
  367. log.Errorf("GetMountTargets for %s error: %v", localFs.Name, err)
  368. return
  369. }
  370. result := localFs.SyncMountTargets(ctx, userCred, mountTargets, xor)
  371. notes := fmt.Sprintf("SyncMountTargets for FileSystem %s result: %s", localFs.Name, result.Result())
  372. log.Infof("%s", notes)
  373. }
  374. func syncVpcPeerConnections(
  375. ctx context.Context,
  376. userCred mcclient.TokenCredential,
  377. syncResults SSyncResultSet,
  378. provider *SCloudprovider,
  379. localVpc *SVpc,
  380. remoteVpc cloudprovider.ICloudVpc,
  381. syncRange *SSyncRange,
  382. ) {
  383. peerConnections, err := func() ([]cloudprovider.ICloudVpcPeeringConnection, error) {
  384. defer syncResults.AddRequestCost(VpcPeeringConnectionManager)()
  385. return remoteVpc.GetICloudVpcPeeringConnections()
  386. }()
  387. if err != nil {
  388. if errors.Cause(err) == cloudprovider.ErrNotImplemented || errors.Cause(err) == cloudprovider.ErrNotSupported {
  389. return
  390. }
  391. log.Errorf("GetICloudVpcPeeringConnections for vpc %s provider %s failed %v", localVpc.Name, provider.Name, err)
  392. return
  393. }
  394. result := func() compare.SyncResult {
  395. defer syncResults.AddSqlCost(VpcPeeringConnectionManager)()
  396. return localVpc.SyncVpcPeeringConnections(ctx, userCred, peerConnections, syncRange.Xor)
  397. }()
  398. syncResults.Add(VpcPeeringConnectionManager, result)
  399. accepterPeerings, err := func() ([]cloudprovider.ICloudVpcPeeringConnection, error) {
  400. defer syncResults.AddRequestCost(VpcPeeringConnectionManager)()
  401. return remoteVpc.GetICloudAccepterVpcPeeringConnections()
  402. }()
  403. if err != nil {
  404. if errors.Cause(err) == cloudprovider.ErrNotImplemented || errors.Cause(err) == cloudprovider.ErrNotSupported {
  405. return
  406. }
  407. log.Errorf("GetICloudVpcPeeringConnections for vpc %s provider %s failed %v", localVpc.Name, provider.Name, err)
  408. return
  409. }
  410. backSyncResult := func() compare.SyncResult {
  411. defer syncResults.AddSqlCost(VpcPeeringConnectionManager)()
  412. return localVpc.BackSycVpcPeeringConnectionsVpc(accepterPeerings)
  413. }()
  414. syncResults.Add(VpcPeeringConnectionManager, backSyncResult)
  415. notes := fmt.Sprintf("SyncVpcPeeringConnections for vpc %s provider %s result: %s", localVpc.Name, provider.Name, result.Result())
  416. log.Infof("%s", notes)
  417. provider.SyncError(result, notes, userCred)
  418. if result.IsError() {
  419. return
  420. }
  421. }
  422. func syncRegionSecGroup(
  423. ctx context.Context,
  424. userCred mcclient.TokenCredential,
  425. syncResults SSyncResultSet,
  426. provider *SCloudprovider,
  427. localRegion *SCloudregion,
  428. remoteRegion cloudprovider.ICloudRegion,
  429. syncRange *SSyncRange,
  430. ) {
  431. secgroups, err := func() ([]cloudprovider.ICloudSecurityGroup, error) {
  432. defer syncResults.AddRequestCost(SecurityGroupManager)()
  433. return remoteRegion.GetISecurityGroups()
  434. }()
  435. if err != nil {
  436. msg := fmt.Sprintf("GetISecurityGroups for region %s provider %s failed %s", localRegion.Name, provider.Name, err)
  437. log.Errorf("%s", msg)
  438. return
  439. }
  440. groups := []cloudprovider.ICloudSecurityGroup{}
  441. for i := range secgroups {
  442. // skip vpc secgroup
  443. if len(secgroups[i].GetVpcId()) > 0 {
  444. continue
  445. }
  446. groups = append(groups, secgroups[i])
  447. }
  448. result := func() compare.SyncResult {
  449. defer syncResults.AddSqlCost(SecurityGroupManager)()
  450. return localRegion.SyncSecgroups(ctx, userCred, provider, nil, groups, syncRange.Xor)
  451. }()
  452. syncResults.Add(SecurityGroupManager, result)
  453. msg := result.Result()
  454. notes := fmt.Sprintf("SyncSecurityGroup for region %s provider %s result: %s", localRegion.Name, provider.Name, msg)
  455. log.Infof("%s", notes)
  456. provider.SyncError(result, notes, userCred)
  457. if result.IsError() {
  458. return
  459. }
  460. }
  461. func syncVpcSecGroup(
  462. ctx context.Context,
  463. userCred mcclient.TokenCredential,
  464. syncResults SSyncResultSet,
  465. provider *SCloudprovider,
  466. localRegion *SCloudregion,
  467. localVpc *SVpc,
  468. remoteVpc cloudprovider.ICloudVpc,
  469. syncRange *SSyncRange,
  470. ) {
  471. secgroups, err := func() ([]cloudprovider.ICloudSecurityGroup, error) {
  472. defer syncResults.AddRequestCost(SecurityGroupManager)()
  473. return remoteVpc.GetISecurityGroups()
  474. }()
  475. if err != nil {
  476. msg := fmt.Sprintf("GetISecurityGroups for vpc %s provider %s failed %s", remoteVpc.GetId(), provider.Name, err)
  477. log.Errorf("%s", msg)
  478. return
  479. }
  480. groups := []cloudprovider.ICloudSecurityGroup{}
  481. for i := range secgroups {
  482. // skip vpc secgroup
  483. if len(secgroups[i].GetVpcId()) == 0 {
  484. continue
  485. }
  486. groups = append(groups, secgroups[i])
  487. }
  488. result := func() compare.SyncResult {
  489. defer syncResults.AddSqlCost(SecurityGroupManager)()
  490. return localRegion.SyncSecgroups(ctx, userCred, provider, localVpc, groups, syncRange.Xor)
  491. }()
  492. syncResults.Add(SecurityGroupManager, result)
  493. msg := result.Result()
  494. notes := fmt.Sprintf("SyncSecurityGroup for VPC %s provider %s result: %s", localVpc.Name, provider.Name, msg)
  495. log.Infof("%s", notes)
  496. provider.SyncError(result, notes, userCred)
  497. if result.IsError() {
  498. return
  499. }
  500. }
  501. func syncVpcRouteTables(ctx context.Context, userCred mcclient.TokenCredential, syncResults SSyncResultSet, provider *SCloudprovider, localVpc *SVpc, remoteVpc cloudprovider.ICloudVpc, syncRange *SSyncRange) {
  502. routeTables, err := func() ([]cloudprovider.ICloudRouteTable, error) {
  503. defer syncResults.AddRequestCost(RouteTableManager)()
  504. return remoteVpc.GetIRouteTables()
  505. }()
  506. if err != nil {
  507. msg := fmt.Sprintf("GetIRouteTables for vpc %s provider %s failed %s", remoteVpc.GetId(), provider.Name, err)
  508. log.Errorf("%s", msg)
  509. return
  510. }
  511. localRouteTables, remoteRouteTables, result := func() ([]SRouteTable, []cloudprovider.ICloudRouteTable, compare.SyncResult) {
  512. defer syncResults.AddSqlCost(RouteTableManager)()
  513. return RouteTableManager.SyncRouteTables(ctx, userCred, localVpc, routeTables, provider, syncRange.Xor)
  514. }()
  515. syncResults.Add(RouteTableManager, result)
  516. msg := result.Result()
  517. notes := fmt.Sprintf("SyncRouteTables for VPC %s provider %s result: %s", localVpc.Name, provider.Name, msg)
  518. log.Infof("%s", notes)
  519. provider.SyncError(result, notes, userCred)
  520. if result.IsError() {
  521. return
  522. }
  523. for i := 0; i < len(localRouteTables); i++ {
  524. func() {
  525. lockman.LockObject(ctx, &localRouteTables[i])
  526. defer lockman.ReleaseObject(ctx, &localRouteTables[i])
  527. if localRouteTables[i].Deleted {
  528. return
  529. }
  530. localRouteTables[i].SyncRouteTableRouteSets(ctx, userCred, remoteRouteTables[i], provider, syncRange.Xor)
  531. localRouteTables[i].SyncRouteTableAssociations(ctx, userCred, remoteRouteTables[i], provider, syncRange.Xor)
  532. }()
  533. }
  534. }
  535. func syncIPv6Gateways(ctx context.Context, userCred mcclient.TokenCredential, syncResults SSyncResultSet, provider *SCloudprovider, localVpc *SVpc, remoteVpc cloudprovider.ICloudVpc, syncRange *SSyncRange) {
  536. exts, err := func() ([]cloudprovider.ICloudIPv6Gateway, error) {
  537. defer syncResults.AddRequestCost(IPv6GatewayManager)()
  538. return remoteVpc.GetICloudIPv6Gateways()
  539. }()
  540. if err != nil {
  541. msg := fmt.Sprintf("GetICloudIPv6Gateways for vpc %s provider %s failed %s", remoteVpc.GetId(), provider.Name, err)
  542. log.Errorf("%s", msg)
  543. return
  544. }
  545. result := func() compare.SyncResult {
  546. defer syncResults.AddSqlCost(IPv6GatewayManager)()
  547. return localVpc.SyncIPv6Gateways(ctx, userCred, exts, provider, syncRange.Xor)
  548. }()
  549. syncResults.Add(IPv6GatewayManager, result)
  550. msg := result.Result()
  551. notes := fmt.Sprintf("SyncIPv6Gateways for VPC %s provider %s result: %s", localVpc.Name, provider.Name, msg)
  552. log.Infof("%s", notes)
  553. provider.SyncError(result, notes, userCred)
  554. if result.IsError() {
  555. return
  556. }
  557. }
  558. func syncVpcNatgateways(ctx context.Context, userCred mcclient.TokenCredential, syncResults SSyncResultSet, provider *SCloudprovider, localVpc *SVpc, remoteVpc cloudprovider.ICloudVpc, syncRange *SSyncRange) {
  559. natGateways, err := func() ([]cloudprovider.ICloudNatGateway, error) {
  560. defer syncResults.AddRequestCost(NatGatewayManager)()
  561. return remoteVpc.GetINatGateways()
  562. }()
  563. if err != nil {
  564. if errors.Cause(err) == cloudprovider.ErrNotImplemented {
  565. return
  566. }
  567. msg := fmt.Sprintf("GetINatGateways for vpc %s provider %s failed %s", remoteVpc.GetId(), provider.Name, err)
  568. log.Errorf("%s", msg)
  569. return
  570. }
  571. localNatGateways, remoteNatGateways, result := func() ([]SNatGateway, []cloudprovider.ICloudNatGateway, compare.SyncResult) {
  572. defer syncResults.AddSqlCost(NatGatewayManager)()
  573. return NatGatewayManager.SyncNatGateways(ctx, userCred, provider.GetOwnerId(), provider, localVpc, natGateways, syncRange.Xor)
  574. }()
  575. syncResults.Add(NatGatewayManager, result)
  576. msg := result.Result()
  577. notes := fmt.Sprintf("SyncNatGateways for VPC %s provider %s result: %s", localVpc.Name, provider.Name, msg)
  578. log.Infof("%s", notes)
  579. provider.SyncError(result, notes, userCred)
  580. if result.IsError() {
  581. return
  582. }
  583. for i := 0; i < len(localNatGateways); i++ {
  584. func() {
  585. lockman.LockObject(ctx, &localNatGateways[i])
  586. defer lockman.ReleaseObject(ctx, &localNatGateways[i])
  587. if localNatGateways[i].Deleted {
  588. return
  589. }
  590. if syncRange.IsNotSkipSyncResource(ElasticipManager) {
  591. syncNatGatewayEips(ctx, userCred, provider, &localNatGateways[i], remoteNatGateways[i])
  592. }
  593. syncNatDTable(ctx, userCred, provider, &localNatGateways[i], remoteNatGateways[i], syncRange.Xor)
  594. syncNatSTable(ctx, userCred, provider, &localNatGateways[i], remoteNatGateways[i], syncRange.Xor)
  595. }()
  596. }
  597. }
  598. func syncNatGatewayEips(ctx context.Context, userCred mcclient.TokenCredential, provider *SCloudprovider, localNatGateway *SNatGateway, remoteNatGateway cloudprovider.ICloudNatGateway) {
  599. eips, err := remoteNatGateway.GetIEips()
  600. if err != nil {
  601. msg := fmt.Sprintf("GetIEIPs for NatGateway %s provider %s failed %s", remoteNatGateway.GetName(), provider.Name, err)
  602. log.Errorf("%s", msg)
  603. return
  604. }
  605. result := localNatGateway.SyncNatGatewayEips(ctx, userCred, provider, eips)
  606. msg := result.Result()
  607. notes := fmt.Sprintf("SyncNatGatewayEips for NatGateway %s provider %s result: %s", localNatGateway.Name, provider.Name, msg)
  608. log.Infof("%s", notes)
  609. provider.SyncError(result, notes, userCred)
  610. if result.IsError() {
  611. return
  612. }
  613. db.OpsLog.LogEvent(provider, db.ACT_SYNC_HOST_COMPLETE, msg, userCred)
  614. }
  615. func syncNatDTable(
  616. ctx context.Context,
  617. userCred mcclient.TokenCredential,
  618. provider *SCloudprovider,
  619. localNatGateway *SNatGateway,
  620. remoteNatGateway cloudprovider.ICloudNatGateway,
  621. xor bool,
  622. ) {
  623. dtable, err := remoteNatGateway.GetINatDTable()
  624. if err != nil {
  625. msg := fmt.Sprintf("GetINatDTable for NatGateway %s provider %s failed %s", remoteNatGateway.GetName(), provider.Name, err)
  626. log.Errorf("%s", msg)
  627. return
  628. }
  629. result := NatDEntryManager.SyncNatDTable(ctx, userCred, provider, localNatGateway, dtable, xor)
  630. msg := result.Result()
  631. notes := fmt.Sprintf("SyncNatDTable for NatGateway %s provider %s result: %s", localNatGateway.Name, provider.Name, msg)
  632. log.Infof("%s", notes)
  633. provider.SyncError(result, notes, userCred)
  634. if result.IsError() {
  635. return
  636. }
  637. db.OpsLog.LogEvent(provider, db.ACT_SYNC_HOST_COMPLETE, msg, userCred)
  638. }
  639. func syncNatSTable(
  640. ctx context.Context,
  641. userCred mcclient.TokenCredential,
  642. provider *SCloudprovider,
  643. localNatGateway *SNatGateway,
  644. remoteNatGateway cloudprovider.ICloudNatGateway,
  645. xor bool,
  646. ) {
  647. stable, err := remoteNatGateway.GetINatSTable()
  648. if err != nil {
  649. msg := fmt.Sprintf("GetINatSTable for NatGateway %s provider %s failed %s", remoteNatGateway.GetName(), provider.Name, err)
  650. log.Errorf("%s", msg)
  651. return
  652. }
  653. result := NatSEntryManager.SyncNatSTable(ctx, userCred, provider, localNatGateway, stable, xor)
  654. msg := result.Result()
  655. notes := fmt.Sprintf("SyncNatSTable for NatGateway %s provider %s result: %s", localNatGateway.Name, provider.Name, msg)
  656. log.Infof("%s", notes)
  657. provider.SyncError(result, notes, userCred)
  658. if result.IsError() {
  659. return
  660. }
  661. db.OpsLog.LogEvent(provider, db.ACT_SYNC_HOST_COMPLETE, msg, userCred)
  662. }
  663. func syncVpcWires(ctx context.Context, userCred mcclient.TokenCredential, syncResults SSyncResultSet, provider *SCloudprovider, localVpc *SVpc, remoteVpc cloudprovider.ICloudVpc, zone *SZone, syncRange *SSyncRange) {
  664. wires, err := func() ([]cloudprovider.ICloudWire, error) {
  665. defer func() {
  666. if syncResults != nil {
  667. syncResults.AddRequestCost(WireManager)()
  668. }
  669. }()
  670. return remoteVpc.GetIWires()
  671. }()
  672. if err != nil {
  673. msg := fmt.Sprintf("GetIWires for vpc %s provider %s failed %s", remoteVpc.GetId(), provider.Name, err)
  674. log.Errorf("%s", msg)
  675. return
  676. }
  677. localWires, remoteWires, result := func() ([]SWire, []cloudprovider.ICloudWire, compare.SyncResult) {
  678. defer func() {
  679. if syncResults != nil {
  680. syncResults.AddSqlCost(WireManager)()
  681. }
  682. }()
  683. return WireManager.SyncWires(ctx, userCred, localVpc, wires, provider, syncRange.Xor, zone)
  684. }()
  685. if syncResults != nil {
  686. syncResults.Add(WireManager, result)
  687. }
  688. msg := result.Result()
  689. notes := fmt.Sprintf("SyncWires for VPC %s provider %s result: %s", localVpc.Name, provider.Name, msg)
  690. log.Infof("%s", notes)
  691. provider.SyncError(result, notes, userCred)
  692. if result.IsError() {
  693. return
  694. }
  695. if localVpc.Id == api.DEFAULT_VPC_ID {
  696. // do not sync on-premise Vpc Network
  697. return
  698. }
  699. for i := 0; i < len(localWires); i += 1 {
  700. func() {
  701. lockman.LockObject(ctx, &localWires[i])
  702. defer lockman.ReleaseObject(ctx, &localWires[i])
  703. if localWires[i].Deleted {
  704. return
  705. }
  706. syncWireNetworks(ctx, userCred, syncResults, provider, &localWires[i], remoteWires[i], syncRange)
  707. }()
  708. }
  709. }
  710. func syncWireNetworks(ctx context.Context, userCred mcclient.TokenCredential, syncResults SSyncResultSet, provider *SCloudprovider, localWire *SWire, remoteWire cloudprovider.ICloudWire, syncRange *SSyncRange) {
  711. nets, err := func() ([]cloudprovider.ICloudNetwork, error) {
  712. defer func() {
  713. if syncResults != nil {
  714. syncResults.AddRequestCost(NetworkManager)()
  715. }
  716. }()
  717. return remoteWire.GetINetworks()
  718. }()
  719. if err != nil {
  720. msg := fmt.Sprintf("GetINetworks for wire %s provider %s failed %s", remoteWire.GetId(), provider.Name, err)
  721. log.Errorf("%s", msg)
  722. return
  723. }
  724. _, _, result := func() ([]SNetwork, []cloudprovider.ICloudNetwork, compare.SyncResult) {
  725. defer func() {
  726. if syncResults != nil {
  727. syncResults.AddSqlCost(NetworkManager)()
  728. }
  729. }()
  730. return NetworkManager.SyncNetworks(ctx, userCred, localWire, nets, provider, syncRange.Xor)
  731. }()
  732. if syncResults != nil {
  733. syncResults.Add(NetworkManager, result)
  734. }
  735. msg := result.Result()
  736. notes := fmt.Sprintf("SyncNetworks for wire %s provider %s result: %s", localWire.Name, provider.Name, msg)
  737. log.Infof("%s", notes)
  738. provider.SyncError(result, notes, userCred)
  739. if result.IsError() {
  740. return
  741. }
  742. }
  743. func syncZoneStorages(
  744. ctx context.Context,
  745. userCred mcclient.TokenCredential,
  746. syncResults SSyncResultSet,
  747. provider *SCloudprovider,
  748. driver cloudprovider.ICloudProvider,
  749. localZone *SZone,
  750. remoteZone cloudprovider.ICloudZone,
  751. syncRange *SSyncRange,
  752. storageCachePairs []sStoragecacheSyncPair,
  753. ) []sStoragecacheSyncPair {
  754. storages, err := func() ([]cloudprovider.ICloudStorage, error) {
  755. defer syncResults.AddRequestCost(StorageManager)()
  756. return remoteZone.GetIStorages()
  757. }()
  758. if err != nil {
  759. msg := fmt.Sprintf("GetIStorages for zone %s provider %s failed %s", remoteZone.GetName(), provider.Name, err)
  760. log.Errorf("%s", msg)
  761. return nil
  762. }
  763. localStorages, remoteStorages, result := func() ([]SStorage, []cloudprovider.ICloudStorage, compare.SyncResult) {
  764. defer syncResults.AddSqlCost(StorageManager)()
  765. return StorageManager.SyncStorages(ctx, userCred, provider, localZone, storages, syncRange.Xor)
  766. }()
  767. syncResults.Add(StorageManager, result)
  768. msg := result.Result()
  769. notes := fmt.Sprintf("SyncStorages for zone %s provider %s result: %s", localZone.Name, provider.Name, msg)
  770. log.Infof("%s", notes)
  771. provider.SyncError(result, notes, userCred)
  772. if result.IsError() {
  773. return nil
  774. }
  775. newCacheIds := make([]sStoragecacheSyncPair, 0)
  776. for i := 0; i < len(localStorages); i += 1 {
  777. func() {
  778. lockman.LockObject(ctx, &localStorages[i])
  779. defer lockman.ReleaseObject(ctx, &localStorages[i])
  780. if localStorages[i].Deleted {
  781. return
  782. }
  783. if !isInCache(storageCachePairs, localStorages[i].StoragecacheId) && !isInCache(newCacheIds, localStorages[i].StoragecacheId) {
  784. cachePair, err := syncStorageCaches(ctx, userCred, provider, &localStorages[i], remoteStorages[i], syncRange.Xor)
  785. if err != nil {
  786. log.Errorf("syncStorageCaches for storage %s(%s) provider %s error: %v", localStorages[i].Name, localStorages[i].Id, provider.Name, err)
  787. }
  788. if cachePair.isValid() {
  789. newCacheIds = append(newCacheIds, cachePair)
  790. }
  791. }
  792. if syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_COMPUTE) {
  793. if !remoteStorages[i].DisableSync() {
  794. syncStorageDisks(ctx, userCred, syncResults, provider, driver, &localStorages[i], remoteStorages[i], syncRange)
  795. }
  796. }
  797. }()
  798. }
  799. return newCacheIds
  800. }
  801. func syncStorageCaches(ctx context.Context, userCred mcclient.TokenCredential, provider *SCloudprovider, localStorage *SStorage, remoteStorage cloudprovider.ICloudStorage, xor bool) (sStoragecacheSyncPair, error) {
  802. var cachePair sStoragecacheSyncPair
  803. log.Debugf("syncStorageCaches for storage %s", localStorage.GetId())
  804. remoteCache := remoteStorage.GetIStoragecache()
  805. if remoteCache == nil {
  806. return cachePair, fmt.Errorf("remote storageCache is nil")
  807. }
  808. localCache, isNew, err := StoragecacheManager.SyncWithCloudStoragecache(ctx, userCred, remoteCache, provider, xor)
  809. if err != nil {
  810. return cachePair, errors.Wrapf(err, "SyncWithCloudStoragecache provider %s with storage %s(%s)", provider.Name, localStorage.Name, localStorage.Id)
  811. }
  812. err = localStorage.SetStoragecache(userCred, localCache)
  813. if err != nil {
  814. return cachePair, errors.Wrapf(err, "SetStoragecache %s(%s)", localCache.Name, localCache.Id)
  815. }
  816. cachePair.local = localCache
  817. cachePair.remote = remoteCache
  818. cachePair.isNew = isNew
  819. cachePair.region, err = localStorage.GetRegion()
  820. if err != nil {
  821. return cachePair, errors.Wrapf(err, "GetRegion for storage %s", localStorage.Id)
  822. }
  823. return cachePair, nil
  824. }
  825. func syncStorageDisks(ctx context.Context, userCred mcclient.TokenCredential, syncResults SSyncResultSet, provider *SCloudprovider, driver cloudprovider.ICloudProvider, localStorage *SStorage, remoteStorage cloudprovider.ICloudStorage, syncRange *SSyncRange) {
  826. disks, err := func() ([]cloudprovider.ICloudDisk, error) {
  827. defer syncResults.AddRequestCost(DiskManager)()
  828. return remoteStorage.GetIDisks()
  829. }()
  830. if err != nil {
  831. msg := fmt.Sprintf("GetIDisks for storage %s provider %s failed %s", remoteStorage.GetName(), provider.Name, err)
  832. log.Errorf("%s", msg)
  833. return
  834. }
  835. _, _, result := func() ([]SDisk, []cloudprovider.ICloudDisk, compare.SyncResult) {
  836. defer syncResults.AddSqlCost(DiskManager)()
  837. return DiskManager.SyncDisks(ctx, userCred, driver, localStorage, disks, provider.GetOwnerId(), syncRange.Xor)
  838. }()
  839. syncResults.Add(DiskManager, result)
  840. msg := result.Result()
  841. notes := fmt.Sprintf("SyncDisks for storage %s provider %s result: %s", localStorage.Name, provider.Name, msg)
  842. log.Infof("%s", notes)
  843. provider.SyncError(result, notes, userCred)
  844. if result.IsError() {
  845. return
  846. }
  847. }
  848. func syncZoneHosts(
  849. ctx context.Context,
  850. userCred mcclient.TokenCredential,
  851. syncResults SSyncResultSet,
  852. provider *SCloudprovider,
  853. driver cloudprovider.ICloudProvider,
  854. localZone *SZone,
  855. remoteZone cloudprovider.ICloudZone,
  856. syncRange *SSyncRange,
  857. storageCachePairs []sStoragecacheSyncPair,
  858. ) []sStoragecacheSyncPair {
  859. hosts, err := func() ([]cloudprovider.ICloudHost, error) {
  860. defer syncResults.AddRequestCost(HostManager)()
  861. return remoteZone.GetIHosts()
  862. }()
  863. if err != nil {
  864. msg := fmt.Sprintf("GetIHosts for zone %s provider %s failed %s", remoteZone.GetName(), provider.Name, err)
  865. log.Errorf("%s", msg)
  866. return nil
  867. }
  868. localHosts, remoteHosts, result := func() ([]SHost, []cloudprovider.ICloudHost, compare.SyncResult) {
  869. defer syncResults.AddSqlCost(HostManager)()
  870. return HostManager.SyncHosts(ctx, userCred, provider, localZone, nil, hosts, syncRange.Xor)
  871. }()
  872. syncResults.Add(HostManager, result)
  873. msg := result.Result()
  874. notes := fmt.Sprintf("SyncHosts for zone %s provider %s result: %s", localZone.Name, provider.Name, msg)
  875. log.Infof("%s", notes)
  876. provider.SyncError(result, notes, userCred)
  877. if result.IsError() {
  878. return nil
  879. }
  880. var newCachePairs []sStoragecacheSyncPair
  881. db.OpsLog.LogEvent(provider, db.ACT_SYNC_HOST_COMPLETE, msg, userCred)
  882. for i := 0; i < len(localHosts); i += 1 {
  883. if len(syncRange.Host) > 0 && !utils.IsInStringArray(localHosts[i].Id, syncRange.Host) {
  884. continue
  885. }
  886. func() {
  887. lockman.LockObject(ctx, &localHosts[i])
  888. defer lockman.ReleaseObject(ctx, &localHosts[i])
  889. if localHosts[i].Deleted {
  890. return
  891. }
  892. newCachePairs = syncHostStorages(ctx, userCred, syncResults, provider, &localHosts[i], remoteHosts[i], storageCachePairs, syncRange.Xor)
  893. if syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_COMPUTE) {
  894. syncHostNics(ctx, userCred, syncResults, provider, &localHosts[i], remoteHosts[i])
  895. // syncHostWires(ctx, userCred, syncResults, provider, &localHosts[i], remoteHosts[i])
  896. syncHostVMs(ctx, userCred, syncResults, provider, driver, &localHosts[i], remoteHosts[i], syncRange)
  897. syncHostIsolateDevices(ctx, userCred, syncResults, provider, driver, &localHosts[i], remoteHosts[i], syncRange)
  898. }
  899. }()
  900. }
  901. return newCachePairs
  902. }
  903. func syncHostStorages(ctx context.Context, userCred mcclient.TokenCredential, syncResults SSyncResultSet, provider *SCloudprovider, localHost *SHost, remoteHost cloudprovider.ICloudHost, storageCachePairs []sStoragecacheSyncPair, xor bool) []sStoragecacheSyncPair {
  904. storages, err := func() ([]cloudprovider.ICloudStorage, error) {
  905. defer syncResults.AddRequestCost(HoststorageManager)()
  906. return remoteHost.GetIStorages()
  907. }()
  908. if err != nil {
  909. msg := fmt.Sprintf("GetIStorages for host %s provider %s failed %s", remoteHost.GetName(), provider.Name, err)
  910. log.Errorf("%s", msg)
  911. return nil
  912. }
  913. localStorages, remoteStorages, result := func() ([]SStorage, []cloudprovider.ICloudStorage, compare.SyncResult) {
  914. defer syncResults.AddSqlCost(HoststorageManager)()
  915. return localHost.SyncHostStorages(ctx, userCred, storages, provider, xor)
  916. }()
  917. syncResults.Add(HoststorageManager, result)
  918. msg := result.Result()
  919. notes := fmt.Sprintf("SyncHostStorages for host %s provider %s result: %s", localHost.Name, provider.Name, msg)
  920. log.Infof("%s", notes)
  921. provider.SyncError(result, notes, userCred)
  922. if result.IsError() {
  923. return nil
  924. }
  925. newCacheIds := make([]sStoragecacheSyncPair, 0)
  926. for i := 0; i < len(localStorages); i += 1 {
  927. syncMetadata(ctx, userCred, &localStorages[i], remoteStorages[i], false)
  928. if !isInCache(storageCachePairs, localStorages[i].StoragecacheId) && !isInCache(newCacheIds, localStorages[i].StoragecacheId) {
  929. cachePair, err := syncStorageCaches(ctx, userCred, provider, &localStorages[i], remoteStorages[i], xor)
  930. if err != nil {
  931. log.Errorf("syncStorageCaches for host %s(%s) error: %v", localHost.Name, localHost.Id, err)
  932. continue
  933. }
  934. if cachePair.remote != nil && cachePair.local != nil {
  935. newCacheIds = append(newCacheIds, cachePair)
  936. }
  937. }
  938. }
  939. return newCacheIds
  940. }
  941. /*func syncHostWires(ctx context.Context, userCred mcclient.TokenCredential, syncResults SSyncResultSet, provider *SCloudprovider, localHost *SHost, remoteHost cloudprovider.ICloudHost) {
  942. netifs, err := func() ([]cloudprovider.ICloudHostNetInterface, error) {
  943. defer func() {
  944. if syncResults != nil {
  945. syncResults.AddRequestCost(NetInterfaceManager)()
  946. }
  947. }()
  948. return remoteHost.GetIHostNics()
  949. }()
  950. if err != nil {
  951. msg := fmt.Sprintf("GetIWires for host %s failed %s", remoteHost.GetName(), err)
  952. log.Errorf("%s", msg)
  953. return
  954. }
  955. result := func() compare.SyncResult {
  956. defer func() {
  957. if syncResults != nil {
  958. syncResults.AddSqlCost(NetInterfaceManager)()
  959. }
  960. }()
  961. return localHost.SyncHostNetInterfaces(ctx, userCred, netifs)
  962. }()
  963. if syncResults != nil {
  964. syncResults.Add(NetInterfaceManager, result)
  965. }
  966. msg := result.Result()
  967. notes := fmt.Sprintf("SyncHostWires/NetInterfaces for host %s result: %s", localHost.Name, msg)
  968. log.Infof("%s", notes)
  969. provider.SyncError(result, notes, userCred)
  970. if result.IsError() {
  971. return
  972. }
  973. db.OpsLog.LogEvent(provider, db.ACT_SYNC_HOST_COMPLETE, msg, userCred)
  974. }*/
  975. func syncHostVMs(ctx context.Context, userCred mcclient.TokenCredential, syncResults SSyncResultSet, provider *SCloudprovider, driver cloudprovider.ICloudProvider, localHost *SHost, remoteHost cloudprovider.ICloudHost, syncRange *SSyncRange) {
  976. vms, err := func() ([]cloudprovider.ICloudVM, error) {
  977. defer syncResults.AddRequestCost(GuestManager)()
  978. return remoteHost.GetIVMs()
  979. }()
  980. if err != nil {
  981. msg := fmt.Sprintf("GetIVMs for host %s provider %s failed %s", remoteHost.GetName(), provider.Name, err)
  982. log.Errorf("%s", msg)
  983. return
  984. }
  985. syncVMPairs, result := func() ([]SGuestSyncResult, compare.SyncResult) {
  986. defer syncResults.AddSqlCost(GuestManager)()
  987. return localHost.SyncHostVMs(ctx, userCred, driver, vms, provider.GetOwnerId(), syncRange.Xor)
  988. }()
  989. syncResults.Add(GuestManager, result)
  990. msg := result.Result()
  991. notes := fmt.Sprintf("SyncHostVMs for host %s provider %s result: %s", localHost.Name, provider.Name, msg)
  992. log.Infof("%s", notes)
  993. provider.SyncError(result, notes, userCred)
  994. for i := 0; i < len(syncVMPairs); i += 1 {
  995. if !syncVMPairs[i].IsNew && !syncRange.DeepSync {
  996. continue
  997. }
  998. func() {
  999. lockman.LockObject(ctx, syncVMPairs[i].Local)
  1000. defer lockman.ReleaseObject(ctx, syncVMPairs[i].Local)
  1001. if syncVMPairs[i].Local.Deleted || syncVMPairs[i].Local.PendingDeleted {
  1002. return
  1003. }
  1004. // 快速同步可能会导致guest被删除,所以需要重新获取
  1005. guest := GuestManager.FetchGuestById(syncVMPairs[i].Local.Id)
  1006. if guest == nil || guest.Deleted || guest.PendingDeleted {
  1007. return
  1008. }
  1009. SyncVMPeripherals(ctx, userCred, syncVMPairs[i].Local, syncVMPairs[i].Remote, localHost, provider, driver)
  1010. }()
  1011. }
  1012. }
  1013. func syncHostIsolateDevices(ctx context.Context, userCred mcclient.TokenCredential, syncResults SSyncResultSet, provider *SCloudprovider, driver cloudprovider.ICloudProvider, localHost *SHost, remoteHost cloudprovider.ICloudHost, syncRange *SSyncRange) {
  1014. devs, err := func() ([]cloudprovider.IsolateDevice, error) {
  1015. defer syncResults.AddRequestCost(HostManager)()
  1016. return remoteHost.GetIsolateDevices()
  1017. }()
  1018. if err != nil {
  1019. if errors.Cause(err) == cloudprovider.ErrNotImplemented || errors.Cause(err) == cloudprovider.ErrNotSupported {
  1020. return
  1021. }
  1022. msg := fmt.Sprintf("GetIsolateDevices for host %s provider %s failed %s", remoteHost.GetName(), provider.Name, err)
  1023. log.Errorf("%s", msg)
  1024. return
  1025. }
  1026. result := func() compare.SyncResult {
  1027. defer syncResults.AddSqlCost(HostManager)()
  1028. return localHost.SyncHostIsolateDevices(ctx, userCred, driver, devs, provider.GetOwnerId(), syncRange.Xor)
  1029. }()
  1030. syncResults.Add(HostManager, result)
  1031. msg := result.Result()
  1032. notes := fmt.Sprintf("SyncHostIsolateDevices for host %s provider %s result: %s", localHost.Name, provider.Name, msg)
  1033. log.Infof("%s", notes)
  1034. provider.SyncError(result, notes, userCred)
  1035. }
  1036. func SyncVMPeripherals(
  1037. ctx context.Context,
  1038. userCred mcclient.TokenCredential,
  1039. local *SGuest,
  1040. remote cloudprovider.ICloudVM,
  1041. host *SHost,
  1042. provider *SCloudprovider,
  1043. driver cloudprovider.ICloudProvider,
  1044. ) {
  1045. err := syncVMNics(ctx, userCred, provider, host, local, remote)
  1046. if err != nil && errors.Cause(err) != cloudprovider.ErrNotSupported && errors.Cause(err) != cloudprovider.ErrNotImplemented {
  1047. logclient.AddSimpleActionLog(local, logclient.ACT_CLOUD_SYNC, errors.Wrapf(err, "syncVMNics"), userCred, false)
  1048. }
  1049. err = syncVMDisks(ctx, userCred, provider, driver, host, local, remote)
  1050. if err != nil && errors.Cause(err) != cloudprovider.ErrNotSupported && errors.Cause(err) != cloudprovider.ErrNotImplemented {
  1051. logclient.AddSimpleActionLog(local, logclient.ACT_CLOUD_SYNC, errors.Wrapf(err, "syncVMDisks"), userCred, false)
  1052. }
  1053. account, _ := provider.GetCloudaccount()
  1054. if account == nil || account.IsNotSkipSyncResource(ElasticipManager) {
  1055. err = syncVMEip(ctx, userCred, provider, local, remote)
  1056. if err != nil &&
  1057. errors.Cause(err) != cloudprovider.ErrNotSupported &&
  1058. errors.Cause(err) != cloudprovider.ErrNotImplemented &&
  1059. errors.Cause(err) != cloudprovider.ErrNotFound {
  1060. logclient.AddSimpleActionLog(local, logclient.ACT_CLOUD_SYNC, errors.Wrapf(err, "syncVMEip"), userCred, false)
  1061. }
  1062. }
  1063. err = syncVMSecgroups(ctx, userCred, provider, local, remote)
  1064. if err != nil && errors.Cause(err) != cloudprovider.ErrNotSupported && errors.Cause(err) != cloudprovider.ErrNotImplemented {
  1065. logclient.AddSimpleActionLog(local, logclient.ACT_CLOUD_SYNC, errors.Wrapf(err, "syncVMSecgroups"), userCred, false)
  1066. }
  1067. result := local.SyncInstanceSnapshots(ctx, userCred, provider)
  1068. if result.IsError() {
  1069. log.Errorf("syncVMInstanceSnapshots error %v", result.AllError())
  1070. }
  1071. err = syncVMIsolateDevices(ctx, userCred, provider, local, remote)
  1072. if err != nil && errors.Cause(err) != cloudprovider.ErrNotSupported && errors.Cause(err) != cloudprovider.ErrNotImplemented {
  1073. logclient.AddSimpleActionLog(local, logclient.ACT_CLOUD_SYNC, errors.Wrapf(err, "syncVMIsolateDevice"), userCred, false)
  1074. }
  1075. err = syncVMContainers(ctx, userCred, provider, local, remote)
  1076. if err != nil && errors.Cause(err) != cloudprovider.ErrNotSupported && errors.Cause(err) != cloudprovider.ErrNotImplemented {
  1077. logclient.AddSimpleActionLog(local, logclient.ACT_CLOUD_SYNC, errors.Wrapf(err, "syncVMContainers"), userCred, false)
  1078. }
  1079. }
  1080. func syncVMContainers(ctx context.Context, userCred mcclient.TokenCredential, provider *SCloudprovider, localVM *SGuest, remoteVM cloudprovider.ICloudVM) error {
  1081. if localVM.Hypervisor != api.HYPERVISOR_POD {
  1082. return nil
  1083. }
  1084. containers, err := remoteVM.GetContainers()
  1085. if err != nil {
  1086. return errors.Wrap(err, "remoteVM.GetContainers")
  1087. }
  1088. result := localVM.SyncVMContainers(ctx, userCred, containers)
  1089. log.Infof("syncVMContainers for VM %s provider %s result: %s", localVM.Name, provider.Name, result.Result())
  1090. if result.IsError() {
  1091. return result.AllError()
  1092. }
  1093. return nil
  1094. }
  1095. func syncVMNics(
  1096. ctx context.Context,
  1097. userCred mcclient.TokenCredential,
  1098. provider *SCloudprovider,
  1099. host *SHost,
  1100. localVM *SGuest,
  1101. remoteVM cloudprovider.ICloudVM,
  1102. ) error {
  1103. nics, err := remoteVM.GetINics()
  1104. if err != nil {
  1105. return errors.Wrap(err, "remoteVM.GetINics")
  1106. }
  1107. result := localVM.SyncVMNics(ctx, userCred, host, nics, nil)
  1108. msg := result.Result()
  1109. notes := fmt.Sprintf("syncVMNics for VM %s provider %s result: %s", localVM.Name, provider.Name, msg)
  1110. log.Infof("%s", notes)
  1111. if result.IsError() {
  1112. return result.AllError()
  1113. }
  1114. return nil
  1115. }
  1116. func syncVMDisks(
  1117. ctx context.Context,
  1118. userCred mcclient.TokenCredential,
  1119. provider *SCloudprovider,
  1120. driver cloudprovider.ICloudProvider,
  1121. host *SHost,
  1122. localVM *SGuest,
  1123. remoteVM cloudprovider.ICloudVM,
  1124. ) error {
  1125. disks, err := remoteVM.GetIDisks()
  1126. if err != nil {
  1127. return errors.Wrap(err, "remoteVM.GetIDisks")
  1128. }
  1129. result := localVM.SyncVMDisks(ctx, userCred, driver, host, disks, provider.GetOwnerId())
  1130. msg := result.Result()
  1131. notes := fmt.Sprintf("syncVMDisks for VM %s provider %s result: %s", localVM.Name, provider.Name, msg)
  1132. log.Infof("%s", notes)
  1133. if result.IsError() {
  1134. return result.AllError()
  1135. }
  1136. return nil
  1137. }
  1138. func syncVMEip(ctx context.Context, userCred mcclient.TokenCredential, provider *SCloudprovider, localVM *SGuest, remoteVM cloudprovider.ICloudVM) error {
  1139. eip, err := remoteVM.GetIEIP()
  1140. if err != nil {
  1141. return errors.Wrap(err, "remoteVM.GetIEIP")
  1142. }
  1143. result := localVM.SyncVMEip(ctx, userCred, provider, eip, provider.GetOwnerId())
  1144. msg := result.Result()
  1145. log.Infof("syncVMEip for VM %s provider %s result: %s", localVM.Name, provider.Name, msg)
  1146. if result.IsError() {
  1147. return result.AllError()
  1148. }
  1149. return nil
  1150. }
  1151. func syncVMSecgroups(ctx context.Context, userCred mcclient.TokenCredential, provider *SCloudprovider, localVM *SGuest, remoteVM cloudprovider.ICloudVM) error {
  1152. secgroupIds, err := remoteVM.GetSecurityGroupIds()
  1153. if err != nil {
  1154. return errors.Wrap(err, "remoteVM.GetSecurityGroupIds")
  1155. }
  1156. return localVM.SyncVMSecgroups(ctx, userCred, secgroupIds)
  1157. }
  1158. func syncVMIsolateDevices(ctx context.Context, userCred mcclient.TokenCredential, provider *SCloudprovider, localVM *SGuest, remoteVM cloudprovider.ICloudVM) error {
  1159. devIds, err := remoteVM.GetIsolateDeviceIds()
  1160. if err != nil {
  1161. return errors.Wrap(err, "remoteVM.GetIsolateDeviceIds")
  1162. }
  1163. return localVM.SyncVMIsolateDevices(ctx, userCred, devIds)
  1164. }
  1165. func (self *SGuest) SyncVMIsolateDevices(ctx context.Context, userCred mcclient.TokenCredential, externalIds []string) error {
  1166. host, err := self.GetHost()
  1167. if err != nil {
  1168. return err
  1169. }
  1170. devs, err := self.GetIsolatedDevices()
  1171. if err != nil {
  1172. return errors.Wrapf(err, "GetIsolatedDevices")
  1173. }
  1174. result := compare.SyncResult{}
  1175. for i := range devs {
  1176. if !utils.IsInStringArray(devs[i].ExternalId, externalIds) {
  1177. _, err = db.Update(&devs[i], func() error {
  1178. devs[i].GuestId = ""
  1179. return nil
  1180. })
  1181. if err != nil {
  1182. return err
  1183. }
  1184. result.Delete()
  1185. }
  1186. }
  1187. devs = []SIsolatedDevice{}
  1188. sq := HostManager.Query("id").Equals("manager_id", host.ManagerId).SubQuery()
  1189. q := IsolatedDeviceManager.Query().In("host_id", sq).In("external_id", externalIds)
  1190. err = db.FetchModelObjects(IsolatedDeviceManager, q, &devs)
  1191. if err != nil {
  1192. return err
  1193. }
  1194. for i := range devs {
  1195. _, err = db.Update(&devs[i], func() error {
  1196. devs[i].GuestId = self.Id
  1197. return nil
  1198. })
  1199. if err != nil {
  1200. return err
  1201. }
  1202. result.Add()
  1203. }
  1204. msg := result.Result()
  1205. notes := fmt.Sprintf("syncHostIsolateDevices for VM %s result: %s", self.Name, msg)
  1206. log.Infof("%s", notes)
  1207. return nil
  1208. }
  1209. func (guest *SGuest) GetContainers() ([]SContainer, error) {
  1210. q := GetContainerManager().Query().Equals("guest_id", guest.Id).Asc("created_at")
  1211. ret := []SContainer{}
  1212. err := db.FetchModelObjects(GetContainerManager(), q, &ret)
  1213. if err != nil {
  1214. return nil, errors.Wrapf(err, "GetContainers")
  1215. }
  1216. return ret, nil
  1217. }
  1218. func (guest *SGuest) SyncVMContainers(
  1219. ctx context.Context,
  1220. userCred mcclient.TokenCredential,
  1221. containers []cloudprovider.ICloudContainer,
  1222. ) compare.SyncResult {
  1223. lockman.LockRawObject(ctx, guest.Id, GetContainerManager().Keyword())
  1224. defer lockman.ReleaseRawObject(ctx, guest.Id, GetContainerManager().Keyword())
  1225. result := compare.SyncResult{}
  1226. dbContainers, err := guest.GetContainers()
  1227. if err != nil {
  1228. result.Error(errors.Wrapf(err, "GetContainers"))
  1229. return result
  1230. }
  1231. removed := make([]SContainer, 0)
  1232. commondb := make([]SContainer, 0)
  1233. commonext := make([]cloudprovider.ICloudContainer, 0)
  1234. added := make([]cloudprovider.ICloudContainer, 0)
  1235. err = compare.CompareSets(dbContainers, containers, &removed, &commondb, &commonext, &added)
  1236. if err != nil {
  1237. result.Error(errors.Wrapf(err, "compare.CompareSets"))
  1238. return result
  1239. }
  1240. for i := 0; i < len(removed); i += 1 {
  1241. err = removed[i].Delete(ctx, userCred)
  1242. if err != nil {
  1243. result.AddError(errors.Wrapf(err, "Delete(%s)", removed[i].Id))
  1244. continue
  1245. }
  1246. result.Delete()
  1247. }
  1248. for i := 0; i < len(commondb); i += 1 {
  1249. if commondb[i].PendingDeleted != guest.PendingDeleted { //避免主机正常,磁盘在回收站的情况
  1250. db.Update(&commondb[i], func() error {
  1251. commondb[i].PendingDeleted = guest.PendingDeleted
  1252. return nil
  1253. })
  1254. }
  1255. err := commondb[i].SyncWithCloudContainer(ctx, userCred, commonext[i])
  1256. if err != nil {
  1257. result.AddError(errors.Wrapf(err, "SyncWithCloudContainer(%s)", commondb[i].Id))
  1258. continue
  1259. }
  1260. commondb[i].SyncCloudProjectId(userCred, guest.GetOwnerId())
  1261. result.Update()
  1262. }
  1263. for i := 0; i < len(added); i += 1 {
  1264. err := guest.syncNewContainer(ctx, userCred, added[i])
  1265. if err != nil {
  1266. result.AddError(errors.Wrapf(err, "syncNewContainer(%s)", added[i].GetGlobalId()))
  1267. continue
  1268. }
  1269. result.Add()
  1270. }
  1271. return result
  1272. }
  1273. func (self *SContainer) SyncWithCloudContainer(ctx context.Context, userCred mcclient.TokenCredential, container cloudprovider.ICloudContainer) error {
  1274. _, err := db.Update(self, func() error {
  1275. self.Status = container.GetStatus()
  1276. self.Spec.Image = container.GetImage()
  1277. self.Spec.Command = container.GetCommand()
  1278. self.Spec.Envs = make([]*apis.ContainerKeyValue, 0)
  1279. for _, env := range container.GetEnvs() {
  1280. self.Spec.Envs = append(self.Spec.Envs, &apis.ContainerKeyValue{
  1281. Key: env.Key,
  1282. Value: env.Value,
  1283. })
  1284. }
  1285. self.Spec.VolumeMounts = make([]*apis.ContainerVolumeMount, 0)
  1286. volumeMounts, err := container.GetVolumentMounts()
  1287. if err != nil && err != cloudprovider.ErrNotImplemented {
  1288. return errors.Wrapf(err, "GetVolumentMounts")
  1289. }
  1290. for _, volumeMount := range volumeMounts {
  1291. self.Spec.VolumeMounts = append(self.Spec.VolumeMounts, &apis.ContainerVolumeMount{
  1292. UniqueName: volumeMount.GetName(),
  1293. ReadOnly: volumeMount.IsReadOnly(),
  1294. Type: apis.ContainerVolumeMountType(volumeMount.GetType()),
  1295. })
  1296. }
  1297. self.Spec.Devices = make([]*api.ContainerDevice, 0)
  1298. devices, err := container.GetDevices()
  1299. if err != nil && err != cloudprovider.ErrNotImplemented {
  1300. return errors.Wrapf(err, "GetDevices")
  1301. }
  1302. for _, device := range devices {
  1303. self.Spec.Devices = append(self.Spec.Devices, &api.ContainerDevice{
  1304. IsolatedDevice: &api.ContainerIsolatedDevice{
  1305. Id: device.GetId(),
  1306. },
  1307. Type: apis.ContainerDeviceType(device.GetType()),
  1308. })
  1309. }
  1310. return nil
  1311. })
  1312. return err
  1313. }
  1314. func (guest *SGuest) syncNewContainer(ctx context.Context, userCred mcclient.TokenCredential, container cloudprovider.ICloudContainer) error {
  1315. res := &SContainer{
  1316. GuestId: guest.Id,
  1317. Spec: &api.ContainerSpec{},
  1318. }
  1319. res.SetModelManager(GetContainerManager(), res)
  1320. res.DomainId = guest.DomainId
  1321. res.ProjectId = guest.ProjectId
  1322. res.StartedAt = container.GetStartedAt()
  1323. res.LastFinishedAt = container.GetLastFinishedAt()
  1324. res.RestartCount = container.GetRestartCount()
  1325. res.ExternalId = container.GetGlobalId()
  1326. res.Name = container.GetName()
  1327. res.Status = container.GetStatus()
  1328. res.Spec.Image = container.GetImage()
  1329. res.Spec.Command = container.GetCommand()
  1330. res.Spec.Envs = make([]*apis.ContainerKeyValue, 0)
  1331. for _, env := range container.GetEnvs() {
  1332. res.Spec.Envs = append(res.Spec.Envs, &apis.ContainerKeyValue{
  1333. Key: env.Key,
  1334. Value: env.Value,
  1335. })
  1336. }
  1337. res.Spec.VolumeMounts = make([]*apis.ContainerVolumeMount, 0)
  1338. volumeMounts, err := container.GetVolumentMounts()
  1339. if err != nil && err != cloudprovider.ErrNotImplemented {
  1340. log.Errorf("GetVolumentMounts error: %v", err)
  1341. }
  1342. for _, volumeMount := range volumeMounts {
  1343. res.Spec.VolumeMounts = append(res.Spec.VolumeMounts, &apis.ContainerVolumeMount{
  1344. UniqueName: volumeMount.GetName(),
  1345. ReadOnly: volumeMount.IsReadOnly(),
  1346. Type: apis.ContainerVolumeMountType(volumeMount.GetType()),
  1347. })
  1348. }
  1349. res.Spec.Devices = make([]*api.ContainerDevice, 0)
  1350. devices, err := container.GetDevices()
  1351. if err != nil && err != cloudprovider.ErrNotImplemented {
  1352. log.Errorf("GetDevices error: %v", err)
  1353. }
  1354. for _, device := range devices {
  1355. res.Spec.Devices = append(res.Spec.Devices, &api.ContainerDevice{
  1356. IsolatedDevice: &api.ContainerIsolatedDevice{
  1357. Id: device.GetId(),
  1358. },
  1359. Type: apis.ContainerDeviceType(device.GetType()),
  1360. })
  1361. }
  1362. err = GetContainerManager().TableSpec().Insert(ctx, res)
  1363. if err != nil {
  1364. return errors.Wrapf(err, "Insert(%s)", res.Id)
  1365. }
  1366. return nil
  1367. }
  1368. func syncSkusFromPrivateCloud(
  1369. ctx context.Context,
  1370. userCred mcclient.TokenCredential,
  1371. syncResults SSyncResultSet,
  1372. region *SCloudregion,
  1373. remoteRegion cloudprovider.ICloudRegion,
  1374. xor bool,
  1375. ) {
  1376. skus, err := remoteRegion.GetISkus()
  1377. if err != nil {
  1378. msg := fmt.Sprintf("GetISkus for region %s(%s) failed %v", region.Name, region.Id, err)
  1379. log.Errorf("%s", msg)
  1380. return
  1381. }
  1382. result := ServerSkuManager.SyncPrivateCloudSkus(ctx, userCred, region, skus, xor)
  1383. syncResults.Add(ServerSkuManager, result)
  1384. msg := result.Result()
  1385. log.Infof("SyncCloudSkusByRegion for region %s result: %s", region.Name, msg)
  1386. if result.IsError() {
  1387. return
  1388. }
  1389. s := auth.GetSession(ctx, userCred, "")
  1390. if _, err := scheduler.SchedManager.SyncSku(s, true); err != nil {
  1391. log.Errorf("Sync scheduler sku cache error: %v", err)
  1392. }
  1393. }
  1394. func syncRegionDBInstances(
  1395. ctx context.Context,
  1396. userCred mcclient.TokenCredential,
  1397. syncResults SSyncResultSet,
  1398. provider *SCloudprovider,
  1399. localRegion *SCloudregion,
  1400. remoteRegion cloudprovider.ICloudRegion,
  1401. syncRange *SSyncRange,
  1402. ) {
  1403. instances, err := func() ([]cloudprovider.ICloudDBInstance, error) {
  1404. defer syncResults.AddRequestCost(DBInstanceManager)()
  1405. return remoteRegion.GetIDBInstances()
  1406. }()
  1407. if err != nil {
  1408. msg := fmt.Sprintf("GetIDBInstances for region %s provider %s failed %s", remoteRegion.GetName(), provider.Name, err)
  1409. log.Errorf("%s", msg)
  1410. return
  1411. }
  1412. localInstances, remoteInstances, result := func() ([]SDBInstance, []cloudprovider.ICloudDBInstance, compare.SyncResult) {
  1413. defer syncResults.AddSqlCost(DBInstanceManager)()
  1414. return DBInstanceManager.SyncDBInstances(ctx, userCred, provider.GetOwnerId(), provider, localRegion, instances, syncRange.Xor)
  1415. }()
  1416. syncResults.Add(DBInstanceManager, result)
  1417. DBInstanceManager.SyncDBInstanceMasterId(ctx, userCred, provider, instances)
  1418. msg := result.Result()
  1419. notes := fmt.Sprintf("SyncDBInstances for region %s provider %s result: %s", localRegion.Name, provider.Name, msg)
  1420. log.Infof("%s", notes)
  1421. provider.SyncError(result, notes, userCred)
  1422. if result.IsError() {
  1423. return
  1424. }
  1425. db.OpsLog.LogEvent(provider, db.ACT_SYNC_HOST_COMPLETE, msg, userCred)
  1426. for i := 0; i < len(localInstances); i++ {
  1427. func() {
  1428. lockman.LockObject(ctx, &localInstances[i])
  1429. defer lockman.ReleaseObject(ctx, &localInstances[i])
  1430. if localInstances[i].Deleted || localInstances[i].PendingDeleted {
  1431. return
  1432. }
  1433. syncDBInstanceResource(ctx, userCred, syncResults, &localInstances[i], remoteInstances[i], syncRange)
  1434. }()
  1435. }
  1436. }
  1437. func syncDBInstanceSkus(ctx context.Context, userCred mcclient.TokenCredential, syncResults SSyncResultSet, provider *SCloudprovider, localRegion *SCloudregion, remoteRegion cloudprovider.ICloudRegion, syncRange *SSyncRange) {
  1438. skus, err := func() ([]cloudprovider.ICloudDBInstanceSku, error) {
  1439. defer syncResults.AddRequestCost(DBInstanceSkuManager)()
  1440. return remoteRegion.GetIDBInstanceSkus()
  1441. }()
  1442. if err != nil {
  1443. if errors.Cause(err) == cloudprovider.ErrNotImplemented {
  1444. return
  1445. }
  1446. msg := fmt.Sprintf("GetIDBInstanceSkus for region %s provider %s failed %s", remoteRegion.GetName(), provider.Name, err)
  1447. log.Errorf("%s", msg)
  1448. return
  1449. }
  1450. result := func() compare.SyncResult {
  1451. defer syncResults.AddSqlCost(DBInstanceSkuManager)()
  1452. return localRegion.SyncDBInstanceSkus(ctx, userCred, provider, skus)
  1453. }()
  1454. syncResults.Add(DBInstanceSkuManager, result)
  1455. msg := result.Result()
  1456. notes := fmt.Sprintf("sync rds sku for region %s provider %s result: %s", localRegion.Name, provider.Name, msg)
  1457. log.Infof("%s", notes)
  1458. provider.SyncError(result, notes, userCred)
  1459. if result.IsError() {
  1460. return
  1461. }
  1462. }
  1463. func syncNATSkus(ctx context.Context, userCred mcclient.TokenCredential, syncResults SSyncResultSet, provider *SCloudprovider, localRegion *SCloudregion, remoteRegion cloudprovider.ICloudRegion, syncRange *SSyncRange) {
  1464. skus, err := func() ([]cloudprovider.ICloudNatSku, error) {
  1465. defer syncResults.AddRequestCost(NatSkuManager)()
  1466. return remoteRegion.GetICloudNatSkus()
  1467. }()
  1468. if err != nil {
  1469. if errors.Cause(err) == cloudprovider.ErrNotImplemented {
  1470. return
  1471. }
  1472. msg := fmt.Sprintf("GetINatSkus for region %s provider %s failed %s", remoteRegion.GetName(), provider.Name, err)
  1473. log.Errorf("%s", msg)
  1474. return
  1475. }
  1476. result := func() compare.SyncResult {
  1477. defer syncResults.AddSqlCost(NatSkuManager)()
  1478. return localRegion.SyncPrivateCloudNatSkus(ctx, userCred, skus)
  1479. }()
  1480. syncResults.Add(NatSkuManager, result)
  1481. msg := result.Result()
  1482. notes := fmt.Sprintf("SyncNatSkus for region %s provider %s result: %s", localRegion.Name, provider.Name, msg)
  1483. log.Infof("%s", notes)
  1484. provider.SyncError(result, notes, userCred)
  1485. if result.IsError() {
  1486. return
  1487. }
  1488. }
  1489. func syncCacheSkus(ctx context.Context, userCred mcclient.TokenCredential, syncResults SSyncResultSet, provider *SCloudprovider, localRegion *SCloudregion, remoteRegion cloudprovider.ICloudRegion, syncRange *SSyncRange) {
  1490. skus, err := func() ([]cloudprovider.ICloudElasticcacheSku, error) {
  1491. defer syncResults.AddRequestCost(ElasticcacheSkuManager)()
  1492. return remoteRegion.GetIElasticcacheSkus()
  1493. }()
  1494. if err != nil {
  1495. if errors.Cause(err) == cloudprovider.ErrNotImplemented {
  1496. return
  1497. }
  1498. msg := fmt.Sprintf("GetIElasticcacheSkus for region %s provider %s failed %s", remoteRegion.GetName(), provider.Name, err)
  1499. log.Errorf("%s", msg)
  1500. return
  1501. }
  1502. result := func() compare.SyncResult {
  1503. defer syncResults.AddSqlCost(ElasticcacheSkuManager)()
  1504. return localRegion.SyncPrivateCloudCacheSkus(ctx, userCred, skus)
  1505. }()
  1506. syncResults.Add(ElasticcacheSkuManager, result)
  1507. msg := result.Result()
  1508. notes := fmt.Sprintf("SyncRedisSkus for region %s provider %s result: %s", localRegion.Name, provider.Name, msg)
  1509. log.Infof("%s", notes)
  1510. provider.SyncError(result, notes, userCred)
  1511. if result.IsError() {
  1512. return
  1513. }
  1514. }
  1515. func syncDBInstanceResource(
  1516. ctx context.Context,
  1517. userCred mcclient.TokenCredential,
  1518. syncResults SSyncResultSet,
  1519. localInstance *SDBInstance,
  1520. remoteInstance cloudprovider.ICloudDBInstance,
  1521. syncRange *SSyncRange,
  1522. ) {
  1523. err := syncDBInstanceNetwork(ctx, userCred, syncResults, localInstance, remoteInstance)
  1524. if err != nil {
  1525. log.Errorf("syncDBInstanceNetwork error: %v", err)
  1526. }
  1527. err = syncDBInstanceSecgroups(ctx, userCred, syncResults, localInstance, remoteInstance)
  1528. if err != nil {
  1529. log.Errorf("syncDBInstanceSecgroups error: %v", err)
  1530. }
  1531. err = syncDBInstanceParameters(ctx, userCred, syncResults, localInstance, remoteInstance)
  1532. if err != nil {
  1533. log.Errorf("syncDBInstanceParameters error: %v", err)
  1534. }
  1535. if syncRange.IsNotSkipSyncResource(DBInstanceDatabaseManager) {
  1536. err = syncDBInstanceDatabases(ctx, userCred, syncResults, localInstance, remoteInstance)
  1537. if err != nil {
  1538. log.Errorf("syncDBInstanceParameters error: %v", err)
  1539. }
  1540. }
  1541. err = syncDBInstanceAccounts(ctx, userCred, syncResults, localInstance, remoteInstance)
  1542. if err != nil {
  1543. log.Errorf("syncDBInstanceAccounts: %v", err)
  1544. }
  1545. if syncRange.IsNotSkipSyncResource(DBInstanceBackupManager) {
  1546. err = syncDBInstanceBackups(ctx, userCred, syncResults, localInstance, remoteInstance)
  1547. if err != nil {
  1548. log.Errorf("syncDBInstanceBackups: %v", err)
  1549. }
  1550. }
  1551. }
  1552. func syncDBInstanceNetwork(ctx context.Context, userCred mcclient.TokenCredential, syncResults SSyncResultSet, localInstance *SDBInstance, remoteInstance cloudprovider.ICloudDBInstance) error {
  1553. networks, err := remoteInstance.GetDBNetworks()
  1554. if err != nil {
  1555. return errors.Wrapf(err, "GetDBNetworks")
  1556. }
  1557. result := DBInstanceNetworkManager.SyncDBInstanceNetwork(ctx, userCred, localInstance, networks)
  1558. syncResults.Add(DBInstanceNetworkManager, result)
  1559. msg := result.Result()
  1560. log.Infof("SyncDBInstanceNetwork for dbinstance %s result: %s", localInstance.Name, msg)
  1561. if result.IsError() {
  1562. return result.AllError()
  1563. }
  1564. return nil
  1565. }
  1566. func syncDBInstanceSecgroups(ctx context.Context, userCred mcclient.TokenCredential, syncResults SSyncResultSet, localInstance *SDBInstance, remoteInstance cloudprovider.ICloudDBInstance) error {
  1567. secIds, err := remoteInstance.GetSecurityGroupIds()
  1568. if err != nil {
  1569. return errors.Wrapf(err, "GetSecurityGroupIds")
  1570. }
  1571. result := DBInstanceSecgroupManager.SyncDBInstanceSecgroups(ctx, userCred, localInstance, secIds)
  1572. syncResults.Add(DBInstanceSecgroupManager, result)
  1573. msg := result.Result()
  1574. log.Infof("SyncDBInstanceSecgroups for dbinstance %s result: %s", localInstance.Name, msg)
  1575. if result.IsError() {
  1576. return result.AllError()
  1577. }
  1578. return nil
  1579. }
  1580. func syncDBInstanceBackups(ctx context.Context, userCred mcclient.TokenCredential, syncResults SSyncResultSet, localInstance *SDBInstance, remoteInstance cloudprovider.ICloudDBInstance) error {
  1581. backups, err := remoteInstance.GetIDBInstanceBackups()
  1582. if err != nil {
  1583. return errors.Wrapf(err, "GetIDBInstanceBackups")
  1584. }
  1585. region, err := localInstance.GetRegion()
  1586. if err != nil {
  1587. return errors.Wrapf(err, "GetRegion")
  1588. }
  1589. provider := localInstance.GetCloudprovider()
  1590. result := DBInstanceBackupManager.SyncDBInstanceBackups(ctx, userCred, provider, localInstance, region, backups, false)
  1591. syncResults.Add(DBInstanceBackupManager, result)
  1592. msg := result.Result()
  1593. log.Infof("SyncDBInstanceBackups for dbinstance %s result: %s", localInstance.Name, msg)
  1594. if result.IsError() {
  1595. return result.AllError()
  1596. }
  1597. return nil
  1598. }
  1599. func syncDBInstanceParameters(ctx context.Context, userCred mcclient.TokenCredential, syncResults SSyncResultSet, localInstance *SDBInstance, remoteInstance cloudprovider.ICloudDBInstance) error {
  1600. parameters, err := remoteInstance.GetIDBInstanceParameters()
  1601. if err != nil {
  1602. return errors.Wrapf(err, "GetIDBInstanceParameters")
  1603. }
  1604. result := DBInstanceParameterManager.SyncDBInstanceParameters(ctx, userCred, localInstance, parameters)
  1605. syncResults.Add(DBInstanceParameterManager, result)
  1606. msg := result.Result()
  1607. log.Infof("SyncDBInstanceParameters for dbinstance %s result: %s", localInstance.Name, msg)
  1608. if result.IsError() {
  1609. return result.AllError()
  1610. }
  1611. return nil
  1612. }
  1613. func syncRegionDBInstanceBackups(
  1614. ctx context.Context,
  1615. userCred mcclient.TokenCredential,
  1616. syncResults SSyncResultSet,
  1617. provider *SCloudprovider,
  1618. localRegion *SCloudregion,
  1619. remoteRegion cloudprovider.ICloudRegion,
  1620. syncRange *SSyncRange,
  1621. ) error {
  1622. backups, err := func() ([]cloudprovider.ICloudDBInstanceBackup, error) {
  1623. defer syncResults.AddRequestCost(DBInstanceBackupManager)()
  1624. return remoteRegion.GetIDBInstanceBackups()
  1625. }()
  1626. if err != nil {
  1627. return errors.Wrapf(err, "GetIDBInstanceBackups")
  1628. }
  1629. result := func() compare.SyncResult {
  1630. defer syncResults.AddSqlCost(DBInstanceBackupManager)()
  1631. return DBInstanceBackupManager.SyncDBInstanceBackups(ctx, userCred, provider, nil, localRegion, backups, syncRange.Xor)
  1632. }()
  1633. syncResults.Add(DBInstanceBackupManager, result)
  1634. msg := result.Result()
  1635. log.Infof("SyncDBInstanceBackups for region %s provider %s result: %s", localRegion.Name, provider.Name, msg)
  1636. if result.IsError() {
  1637. return result.AllError()
  1638. }
  1639. return nil
  1640. }
  1641. func syncDBInstanceDatabases(ctx context.Context, userCred mcclient.TokenCredential, syncResults SSyncResultSet, localInstance *SDBInstance, remoteInstance cloudprovider.ICloudDBInstance) error {
  1642. databases, err := remoteInstance.GetIDBInstanceDatabases()
  1643. if err != nil {
  1644. return errors.Wrapf(err, "GetIDBInstanceDatabases")
  1645. }
  1646. result := DBInstanceDatabaseManager.SyncDBInstanceDatabases(ctx, userCred, localInstance, databases)
  1647. syncResults.Add(DBInstanceDatabaseManager, result)
  1648. msg := result.Result()
  1649. log.Infof("SyncDBInstanceDatabases for dbinstance %s result: %s", localInstance.Name, msg)
  1650. if result.IsError() {
  1651. return result.AllError()
  1652. }
  1653. return nil
  1654. }
  1655. func syncDBInstanceAccounts(ctx context.Context, userCred mcclient.TokenCredential, syncResults SSyncResultSet, localInstance *SDBInstance, remoteInstance cloudprovider.ICloudDBInstance) error {
  1656. accounts, err := remoteInstance.GetIDBInstanceAccounts()
  1657. if err != nil {
  1658. return errors.Wrapf(err, "GetIDBInstanceAccounts")
  1659. }
  1660. localAccounts, remoteAccounts, result := DBInstanceAccountManager.SyncDBInstanceAccounts(ctx, userCred, localInstance, accounts)
  1661. syncResults.Add(DBInstanceDatabaseManager, result)
  1662. msg := result.Result()
  1663. log.Infof("SyncDBInstanceAccounts for dbinstance %s result: %s", localInstance.Name, msg)
  1664. if result.IsError() {
  1665. return result.AllError()
  1666. }
  1667. for i := 0; i < len(localAccounts); i++ {
  1668. func() {
  1669. lockman.LockObject(ctx, &localAccounts[i])
  1670. defer lockman.ReleaseObject(ctx, &localAccounts[i])
  1671. if localAccounts[i].Deleted {
  1672. return
  1673. }
  1674. err = syncDBInstanceAccountPrivileges(ctx, userCred, syncResults, &localAccounts[i], remoteAccounts[i])
  1675. if err != nil {
  1676. log.Errorf("syncDBInstanceAccountPrivileges error: %v", err)
  1677. }
  1678. }()
  1679. }
  1680. return nil
  1681. }
  1682. func syncDBInstanceAccountPrivileges(ctx context.Context, userCred mcclient.TokenCredential, syncResults SSyncResultSet, localAccount *SDBInstanceAccount, remoteAccount cloudprovider.ICloudDBInstanceAccount) error {
  1683. privileges, err := remoteAccount.GetIDBInstanceAccountPrivileges()
  1684. if err != nil {
  1685. return errors.Wrapf(err, "GetIDBInstanceAccountPrivileges for %s(%s)", localAccount.Name, localAccount.Id)
  1686. }
  1687. result := DBInstancePrivilegeManager.SyncDBInstanceAccountPrivileges(ctx, userCred, localAccount, privileges)
  1688. syncResults.Add(DBInstancePrivilegeManager, result)
  1689. msg := result.Result()
  1690. log.Infof("SyncDBInstanceAccountPrivileges for account %s result: %s", localAccount.Name, msg)
  1691. if result.IsError() {
  1692. return result.AllError()
  1693. }
  1694. return nil
  1695. }
  1696. func syncWafIPSets(
  1697. ctx context.Context,
  1698. userCred mcclient.TokenCredential,
  1699. syncResults SSyncResultSet,
  1700. provider *SCloudprovider,
  1701. localRegion *SCloudregion,
  1702. remoteRegion cloudprovider.ICloudRegion,
  1703. xor bool,
  1704. ) error {
  1705. ipSets, err := func() ([]cloudprovider.ICloudWafIPSet, error) {
  1706. defer syncResults.AddRequestCost(WafIPSetManager)()
  1707. return remoteRegion.GetICloudWafIPSets()
  1708. }()
  1709. if err != nil {
  1710. if errors.Cause(err) == cloudprovider.ErrNotImplemented || errors.Cause(err) == cloudprovider.ErrNotSupported {
  1711. return nil
  1712. }
  1713. msg := fmt.Sprintf("GetICloudWafIPSets for region %s provider %s failed %s", remoteRegion.GetName(), provider.Name, err)
  1714. log.Errorf("%s", msg)
  1715. return err
  1716. }
  1717. result := func() compare.SyncResult {
  1718. defer syncResults.AddSqlCost(WafIPSetManager)()
  1719. return localRegion.SyncWafIPSets(ctx, userCred, provider, ipSets, xor)
  1720. }()
  1721. syncResults.Add(WafIPSetManager, result)
  1722. notes := fmt.Sprintf("SyncWafIPSets for region %s provider %s result: %s", localRegion.Name, provider.Name, result.Result())
  1723. log.Infof("%s", notes)
  1724. provider.SyncError(result, notes, userCred)
  1725. if result.IsError() {
  1726. return result.AllError()
  1727. }
  1728. return nil
  1729. }
  1730. func syncWafRegexSets(
  1731. ctx context.Context,
  1732. userCred mcclient.TokenCredential,
  1733. syncResults SSyncResultSet,
  1734. provider *SCloudprovider,
  1735. localRegion *SCloudregion,
  1736. remoteRegion cloudprovider.ICloudRegion,
  1737. xor bool,
  1738. ) error {
  1739. rSets, err := func() ([]cloudprovider.ICloudWafRegexSet, error) {
  1740. defer syncResults.AddRequestCost(WafRegexSetManager)()
  1741. return remoteRegion.GetICloudWafRegexSets()
  1742. }()
  1743. if err != nil {
  1744. if errors.Cause(err) == cloudprovider.ErrNotImplemented || errors.Cause(err) == cloudprovider.ErrNotSupported {
  1745. return nil
  1746. }
  1747. msg := fmt.Sprintf("GetICloudWafRegexSets for region %s provider %s failed %s", remoteRegion.GetName(), provider.Name, err)
  1748. log.Errorf("%s", msg)
  1749. return err
  1750. }
  1751. result := func() compare.SyncResult {
  1752. defer syncResults.AddSqlCost(WafRegexSetManager)()
  1753. return localRegion.SyncWafRegexSets(ctx, userCred, provider, rSets, xor)
  1754. }()
  1755. syncResults.Add(WafRegexSetManager, result)
  1756. notes := fmt.Sprintf("SyncWafRegexSets for region %s provider %s result: %s", localRegion.Name, provider.Name, result.Result())
  1757. log.Infof("%s", notes)
  1758. provider.SyncError(result, notes, userCred)
  1759. if result.IsError() {
  1760. return result.AllError()
  1761. }
  1762. return nil
  1763. }
  1764. func syncMongoDBs(
  1765. ctx context.Context,
  1766. userCred mcclient.TokenCredential,
  1767. syncResults SSyncResultSet,
  1768. provider *SCloudprovider,
  1769. localRegion *SCloudregion,
  1770. remoteRegion cloudprovider.ICloudRegion,
  1771. xor bool,
  1772. ) error {
  1773. dbs, err := func() ([]cloudprovider.ICloudMongoDB, error) {
  1774. defer syncResults.AddRequestCost(MongoDBManager)()
  1775. return remoteRegion.GetICloudMongoDBs()
  1776. }()
  1777. if err != nil {
  1778. msg := fmt.Sprintf("GetICloudMongoDBs for region %s provider %s failed %s", remoteRegion.GetName(), provider.Name, err)
  1779. log.Errorf("%s", msg)
  1780. return err
  1781. }
  1782. _, _, result := func() ([]SMongoDB, []cloudprovider.ICloudMongoDB, compare.SyncResult) {
  1783. defer syncResults.AddSqlCost(MongoDBManager)()
  1784. return localRegion.SyncMongoDBs(ctx, userCred, provider, dbs, xor)
  1785. }()
  1786. syncResults.Add(MongoDBManager, result)
  1787. msg := result.Result()
  1788. notes := fmt.Sprintf("SyncMongoDBs for region %s provider %s result: %s", localRegion.Name, provider.Name, msg)
  1789. log.Infof("%s", notes)
  1790. provider.SyncError(result, notes, userCred)
  1791. if result.IsError() {
  1792. return result.AllError()
  1793. }
  1794. return nil
  1795. }
  1796. func syncElasticSearchs(
  1797. ctx context.Context,
  1798. userCred mcclient.TokenCredential,
  1799. syncResults SSyncResultSet,
  1800. provider *SCloudprovider,
  1801. localRegion *SCloudregion,
  1802. remoteRegion cloudprovider.ICloudRegion,
  1803. xor bool,
  1804. ) error {
  1805. iEss, err := func() ([]cloudprovider.ICloudElasticSearch, error) {
  1806. defer syncResults.AddRequestCost(ElasticSearchManager)()
  1807. return remoteRegion.GetIElasticSearchs()
  1808. }()
  1809. if err != nil {
  1810. msg := fmt.Sprintf("GetIElasticSearchs for region %s provider %s failed %s", remoteRegion.GetName(), provider.Name, err)
  1811. log.Errorf("%s", msg)
  1812. return err
  1813. }
  1814. result := func() compare.SyncResult {
  1815. defer syncResults.AddSqlCost(ElasticSearchManager)()
  1816. return localRegion.SyncElasticSearchs(ctx, userCred, provider, iEss, xor)
  1817. }()
  1818. syncResults.Add(ElasticSearchManager, result)
  1819. msg := result.Result()
  1820. notes := fmt.Sprintf("SyncElasticSearchs for region %s provider %s result: %s", localRegion.Name, provider.Name, msg)
  1821. log.Infof("%s", notes)
  1822. provider.SyncError(result, notes, userCred)
  1823. if result.IsError() {
  1824. return result.AllError()
  1825. }
  1826. return nil
  1827. }
  1828. func syncKafkas(
  1829. ctx context.Context,
  1830. userCred mcclient.TokenCredential,
  1831. syncResults SSyncResultSet,
  1832. provider *SCloudprovider,
  1833. localRegion *SCloudregion,
  1834. remoteRegion cloudprovider.ICloudRegion,
  1835. xor bool,
  1836. ) error {
  1837. iKafkas, err := func() ([]cloudprovider.ICloudKafka, error) {
  1838. defer syncResults.AddRequestCost(KafkaManager)()
  1839. return remoteRegion.GetICloudKafkas()
  1840. }()
  1841. if err != nil {
  1842. msg := fmt.Sprintf("GetICloudKafkas for region %s provider %s failed %s", remoteRegion.GetName(), provider.Name, err)
  1843. log.Errorf("%s", msg)
  1844. return err
  1845. }
  1846. result := func() compare.SyncResult {
  1847. defer syncResults.AddSqlCost(KafkaManager)()
  1848. return localRegion.SyncKafkas(ctx, userCred, provider, iKafkas, xor)
  1849. }()
  1850. syncResults.Add(KafkaManager, result)
  1851. msg := result.Result()
  1852. notes := fmt.Sprintf("SyncKafkas for region %s provider %s result: %s", localRegion.Name, provider.Name, msg)
  1853. log.Infof("%s", notes)
  1854. provider.SyncError(result, notes, userCred)
  1855. if result.IsError() {
  1856. return result.AllError()
  1857. }
  1858. return nil
  1859. }
  1860. func syncApps(
  1861. ctx context.Context,
  1862. userCred mcclient.TokenCredential,
  1863. syncResults SSyncResultSet,
  1864. provider *SCloudprovider,
  1865. localRegion *SCloudregion,
  1866. remoteRegion cloudprovider.ICloudRegion,
  1867. xor bool,
  1868. ) error {
  1869. iApps, err := func() ([]cloudprovider.ICloudApp, error) {
  1870. defer syncResults.AddRequestCost(AppManager)()
  1871. return remoteRegion.GetICloudApps()
  1872. }()
  1873. if err != nil {
  1874. msg := fmt.Sprintf("GetICloudApps for region %s provider %s failed %s", remoteRegion.GetName(), provider.Name, err)
  1875. log.Errorf("%s", msg)
  1876. return err
  1877. }
  1878. result := func() compare.SyncResult {
  1879. defer syncResults.AddSqlCost(AppManager)()
  1880. return localRegion.SyncApps(ctx, userCred, provider, iApps, xor)
  1881. }()
  1882. syncResults.Add(AppManager, result)
  1883. msg := result.Result()
  1884. notes := fmt.Sprintf("SyncApps for region %s provider %s result: %s", localRegion.Name, provider.Name, msg)
  1885. log.Infof("%s", notes)
  1886. provider.SyncError(result, notes, userCred)
  1887. if result.IsError() {
  1888. return result.AllError()
  1889. }
  1890. return nil
  1891. }
  1892. func syncKubeClusters(
  1893. ctx context.Context,
  1894. userCred mcclient.TokenCredential,
  1895. syncResults SSyncResultSet,
  1896. provider *SCloudprovider,
  1897. localRegion *SCloudregion,
  1898. remoteRegion cloudprovider.ICloudRegion,
  1899. xor bool,
  1900. ) error {
  1901. iClusters, err := func() ([]cloudprovider.ICloudKubeCluster, error) {
  1902. defer syncResults.AddRequestCost(KubeClusterManager)()
  1903. return remoteRegion.GetICloudKubeClusters()
  1904. }()
  1905. if err != nil {
  1906. msg := fmt.Sprintf("GetICloudKubeClusters for region %s provider %s failed %s", remoteRegion.GetName(), provider.Name, err)
  1907. log.Errorf("%s", msg)
  1908. return err
  1909. }
  1910. localClusters, remoteClusters, result := func() ([]SKubeCluster, []cloudprovider.ICloudKubeCluster, compare.SyncResult) {
  1911. defer syncResults.AddSqlCost(KubeClusterManager)()
  1912. return localRegion.SyncKubeClusters(ctx, userCred, provider, iClusters, xor)
  1913. }()
  1914. syncResults.Add(KubeClusterManager, result)
  1915. msg := result.Result()
  1916. notes := fmt.Sprintf("SyncKubeClusters for region %s provider %s result: %s", localRegion.Name, provider.Name, msg)
  1917. log.Infof("%s", notes)
  1918. provider.SyncError(result, notes, userCred)
  1919. if result.IsError() {
  1920. return result.AllError()
  1921. }
  1922. for i := 0; i < len(localClusters); i++ {
  1923. func() {
  1924. lockman.LockObject(ctx, &localClusters[i])
  1925. defer lockman.ReleaseObject(ctx, &localClusters[i])
  1926. if localClusters[i].Deleted {
  1927. return
  1928. }
  1929. if err := syncKubeClusterNodePools(ctx, userCred, syncResults, &localClusters[i], remoteClusters[i]); err != nil {
  1930. log.Errorf("syncKubeClusterNodePools for %s error: %v", localClusters[i].Name, err)
  1931. }
  1932. if err := syncKubeClusterNodes(ctx, userCred, syncResults, &localClusters[i], remoteClusters[i]); err != nil {
  1933. log.Errorf("syncKubeClusterNodes for %s error: %v", localClusters[i].Name, err)
  1934. }
  1935. if err := localClusters[i].ImportOrUpdate(ctx, userCred, remoteClusters[i]); err != nil {
  1936. log.Errorf("Import cluster %s error: %v", localClusters[i].Name, err)
  1937. }
  1938. }()
  1939. }
  1940. return nil
  1941. }
  1942. func syncKubeClusterNodePools(ctx context.Context, userCred mcclient.TokenCredential, syncResults SSyncResultSet, cluster *SKubeCluster, ext cloudprovider.ICloudKubeCluster) error {
  1943. iPools, err := func() ([]cloudprovider.ICloudKubeNodePool, error) {
  1944. defer syncResults.AddRequestCost(KubeNodePoolManager)()
  1945. return ext.GetIKubeNodePools()
  1946. }()
  1947. if err != nil {
  1948. msg := fmt.Sprintf("GetICloudKubeNodePools for cluster %s failed %s", cluster.GetName(), err)
  1949. log.Errorf("%s", msg)
  1950. return err
  1951. }
  1952. result := func() compare.SyncResult {
  1953. defer syncResults.AddSqlCost(KubeNodePoolManager)()
  1954. return cluster.SyncKubeNodePools(ctx, userCred, iPools)
  1955. }()
  1956. syncResults.Add(KubeNodePoolManager, result)
  1957. msg := result.Result()
  1958. notes := fmt.Sprintf("SyncKubeNodePools for cluster %s result: %s", cluster.Name, msg)
  1959. log.Infof("%s", notes)
  1960. if result.IsError() {
  1961. return result.AllError()
  1962. }
  1963. return nil
  1964. }
  1965. func syncKubeClusterNodes(ctx context.Context, userCred mcclient.TokenCredential, syncResults SSyncResultSet, cluster *SKubeCluster, ext cloudprovider.ICloudKubeCluster) error {
  1966. iNodes, err := func() ([]cloudprovider.ICloudKubeNode, error) {
  1967. defer syncResults.AddRequestCost(KubeNodeManager)()
  1968. return ext.GetIKubeNodes()
  1969. }()
  1970. if err != nil {
  1971. msg := fmt.Sprintf("GetICloudKubeNodes for cluster %s failed %s", cluster.GetName(), err)
  1972. log.Errorf("%s", msg)
  1973. return err
  1974. }
  1975. result := func() compare.SyncResult {
  1976. defer syncResults.AddSqlCost(KubeNodeManager)()
  1977. return cluster.SyncKubeNodes(ctx, userCred, iNodes)
  1978. }()
  1979. syncResults.Add(KubeNodeManager, result)
  1980. notes := fmt.Sprintf("SyncKubeNodes for cluster %s result: %s", cluster.Name, result.Result())
  1981. log.Infof("%s", notes)
  1982. if result.IsError() {
  1983. return result.AllError()
  1984. }
  1985. return nil
  1986. }
  1987. func syncWafInstances(
  1988. ctx context.Context,
  1989. userCred mcclient.TokenCredential,
  1990. syncResults SSyncResultSet,
  1991. provider *SCloudprovider,
  1992. localRegion *SCloudregion,
  1993. remoteRegion cloudprovider.ICloudRegion,
  1994. xor bool,
  1995. ) error {
  1996. wafIns, err := func() ([]cloudprovider.ICloudWafInstance, error) {
  1997. defer syncResults.AddRequestCost(WafInstanceManager)()
  1998. return remoteRegion.GetICloudWafInstances()
  1999. }()
  2000. if err != nil {
  2001. msg := fmt.Sprintf("GetICloudWafInstances for region %s provider %s failed %s", remoteRegion.GetName(), provider.Name, err)
  2002. log.Errorf("%s", msg)
  2003. return err
  2004. }
  2005. localWafs, remoteWafs, result := func() ([]SWafInstance, []cloudprovider.ICloudWafInstance, compare.SyncResult) {
  2006. defer syncResults.AddSqlCost(WafInstanceManager)()
  2007. return localRegion.SyncWafInstances(ctx, userCred, provider, wafIns, xor)
  2008. }()
  2009. syncResults.Add(WafInstanceManager, result)
  2010. msg := result.Result()
  2011. notes := fmt.Sprintf("SyncWafInstances for region %s provider %s result: %s", localRegion.Name, provider.Name, msg)
  2012. log.Infof("%s", notes)
  2013. provider.SyncError(result, notes, userCred)
  2014. if result.IsError() {
  2015. return result.AllError()
  2016. }
  2017. for i := 0; i < len(localWafs); i++ {
  2018. func() {
  2019. lockman.LockObject(ctx, &localWafs[i])
  2020. defer lockman.ReleaseObject(ctx, &localWafs[i])
  2021. if localWafs[i].Deleted {
  2022. return
  2023. }
  2024. err = syncWafRules(ctx, userCred, syncResults, &localWafs[i], remoteWafs[i])
  2025. if err != nil {
  2026. log.Errorf("syncWafRules error: %v", err)
  2027. }
  2028. }()
  2029. }
  2030. return nil
  2031. }
  2032. func syncWafRules(ctx context.Context, userCred mcclient.TokenCredential, syncResults SSyncResultSet, localWaf *SWafInstance, remoteWafs cloudprovider.ICloudWafInstance) error {
  2033. rules, err := func() ([]cloudprovider.ICloudWafRule, error) {
  2034. defer syncResults.AddRequestCost(WafRuleManager)()
  2035. return remoteWafs.GetRules()
  2036. }()
  2037. if err != nil {
  2038. if errors.Cause(err) == cloudprovider.ErrNotImplemented || errors.Cause(err) == cloudprovider.ErrNotSupported {
  2039. return nil
  2040. }
  2041. msg := fmt.Sprintf("GetRules for waf instance %s failed %s", localWaf.Name, err)
  2042. log.Errorf("%s", msg)
  2043. return err
  2044. }
  2045. result := func() compare.SyncResult {
  2046. defer syncResults.AddSqlCost(WafRuleManager)()
  2047. return localWaf.SyncWafRules(ctx, userCred, rules)
  2048. }()
  2049. syncResults.Add(WafRuleManager, result)
  2050. msg := result.Result()
  2051. notes := fmt.Sprintf("SyncWafRules for waf %s result: %s", localWaf.Name, msg)
  2052. log.Infof("%s", notes)
  2053. if result.IsError() {
  2054. return result.AllError()
  2055. }
  2056. return nil
  2057. }
  2058. func syncRegionSnapshots(
  2059. ctx context.Context,
  2060. userCred mcclient.TokenCredential,
  2061. syncResults SSyncResultSet,
  2062. provider *SCloudprovider,
  2063. localRegion *SCloudregion,
  2064. remoteRegion cloudprovider.ICloudRegion,
  2065. syncRange *SSyncRange,
  2066. ) {
  2067. snapshots, err := func() ([]cloudprovider.ICloudSnapshot, error) {
  2068. defer syncResults.AddRequestCost(SnapshotManager)()
  2069. return remoteRegion.GetISnapshots()
  2070. }()
  2071. if err != nil {
  2072. if errors.Cause(err) == cloudprovider.ErrNotImplemented || errors.Cause(err) == cloudprovider.ErrNotSupported {
  2073. return
  2074. }
  2075. msg := fmt.Sprintf("GetISnapshots for region %s provider %s failed %s", remoteRegion.GetName(), provider.Name, err)
  2076. log.Errorf("%s", msg)
  2077. return
  2078. }
  2079. result := func() compare.SyncResult {
  2080. defer syncResults.AddSqlCost(SnapshotManager)()
  2081. return SnapshotManager.SyncSnapshots(ctx, userCred, provider, localRegion, snapshots, provider.GetOwnerId(), syncRange.Xor)
  2082. }()
  2083. syncResults.Add(SnapshotManager, result)
  2084. msg := result.Result()
  2085. notes := fmt.Sprintf("SyncSnapshots for region %s provider %s result: %s", localRegion.Name, provider.Name, msg)
  2086. log.Infof("%s", notes)
  2087. provider.SyncError(result, notes, userCred)
  2088. if result.IsError() {
  2089. return
  2090. }
  2091. }
  2092. func syncRegionSnapshotPolicies(
  2093. ctx context.Context,
  2094. userCred mcclient.TokenCredential,
  2095. syncResults SSyncResultSet,
  2096. provider *SCloudprovider,
  2097. localRegion *SCloudregion,
  2098. remoteRegion cloudprovider.ICloudRegion,
  2099. syncRange *SSyncRange,
  2100. ) {
  2101. snapshotPolicies, err := func() ([]cloudprovider.ICloudSnapshotPolicy, error) {
  2102. defer syncResults.AddRequestCost(SnapshotPolicyManager)()
  2103. return remoteRegion.GetISnapshotPolicies()
  2104. }()
  2105. if err != nil {
  2106. if errors.Cause(err) == cloudprovider.ErrNotImplemented || errors.Cause(err) == cloudprovider.ErrNotSupported {
  2107. return
  2108. }
  2109. msg := fmt.Sprintf("GetISnapshotPolicies for region %s provider %s failed %s", remoteRegion.GetName(), provider.Name, err)
  2110. log.Errorf("%s", msg)
  2111. return
  2112. }
  2113. result := func() compare.SyncResult {
  2114. defer syncResults.AddSqlCost(SnapshotPolicyManager)()
  2115. return localRegion.SyncSnapshotPolicies(ctx, userCred, provider, snapshotPolicies, provider.GetOwnerId(), syncRange.Xor)
  2116. }()
  2117. syncResults.Add(SnapshotPolicyManager, result)
  2118. msg := result.Result()
  2119. notes := fmt.Sprintf("SyncSnapshotPolicies for region %s provider %s result: %s", localRegion.Name, provider.Name, msg)
  2120. log.Infof("%s", notes)
  2121. provider.SyncError(result, notes, userCred)
  2122. if result.IsError() {
  2123. return
  2124. }
  2125. }
  2126. func syncRegionNetworkInterfaces(
  2127. ctx context.Context,
  2128. userCred mcclient.TokenCredential,
  2129. syncResults SSyncResultSet,
  2130. provider *SCloudprovider,
  2131. localRegion *SCloudregion,
  2132. remoteRegion cloudprovider.ICloudRegion,
  2133. syncRange *SSyncRange,
  2134. ) {
  2135. networkInterfaces, err := func() ([]cloudprovider.ICloudNetworkInterface, error) {
  2136. defer syncResults.AddRequestCost(NetworkInterfaceManager)()
  2137. return remoteRegion.GetINetworkInterfaces()
  2138. }()
  2139. if err != nil {
  2140. if errors.Cause(err) == cloudprovider.ErrNotImplemented || errors.Cause(err) == cloudprovider.ErrNotSupported {
  2141. return
  2142. }
  2143. msg := fmt.Sprintf("GetINetworkInterfaces for region %s provider %s failed %s", remoteRegion.GetName(), provider.Name, err)
  2144. log.Errorf("%s", msg)
  2145. return
  2146. }
  2147. localInterfaces, remoteInterfaces, result := func() ([]SNetworkInterface, []cloudprovider.ICloudNetworkInterface, compare.SyncResult) {
  2148. defer syncResults.AddSqlCost(NetworkInterfaceManager)()
  2149. return NetworkInterfaceManager.SyncNetworkInterfaces(ctx, userCred, provider, localRegion, networkInterfaces, syncRange.Xor)
  2150. }()
  2151. syncResults.Add(NetworkInterfaceManager, result)
  2152. msg := result.Result()
  2153. notes := fmt.Sprintf("SyncNetworkInterfaces for region %s provider %s result: %s", localRegion.Name, provider.Name, msg)
  2154. log.Infof("%s", notes)
  2155. provider.SyncError(result, notes, userCred)
  2156. if result.IsError() {
  2157. return
  2158. }
  2159. for i := 0; i < len(localInterfaces); i++ {
  2160. func() {
  2161. lockman.LockObject(ctx, &localInterfaces[i])
  2162. defer lockman.ReleaseObject(ctx, &localInterfaces[i])
  2163. if localInterfaces[i].Deleted {
  2164. return
  2165. }
  2166. syncInterfaceAddresses(ctx, userCred, provider, &localInterfaces[i], remoteInterfaces[i])
  2167. }()
  2168. }
  2169. }
  2170. func syncInterfaceAddresses(ctx context.Context, userCred mcclient.TokenCredential, provider *SCloudprovider, localInterface *SNetworkInterface, remoteInterface cloudprovider.ICloudNetworkInterface) {
  2171. addresses, err := remoteInterface.GetICloudInterfaceAddresses()
  2172. if err != nil {
  2173. if errors.Cause(err) == cloudprovider.ErrNotImplemented || errors.Cause(err) == cloudprovider.ErrNotSupported {
  2174. return
  2175. }
  2176. msg := fmt.Sprintf("GetICloudInterfaceAddresses for networkinterface %s provider %s failed %s", remoteInterface.GetName(), provider.Name, err)
  2177. log.Errorf("%s", msg)
  2178. return
  2179. }
  2180. result := NetworkinterfacenetworkManager.SyncInterfaceAddresses(ctx, userCred, localInterface, addresses)
  2181. msg := result.Result()
  2182. notes := fmt.Sprintf("SyncInterfaceAddresses for networkinterface %s provider %s result: %s", localInterface.Name, provider.Name, msg)
  2183. log.Infof("%s", notes)
  2184. if result.IsError() {
  2185. return
  2186. }
  2187. }
  2188. func syncPublicCloudProviderInfo(
  2189. ctx context.Context,
  2190. userCred mcclient.TokenCredential,
  2191. syncResults SSyncResultSet,
  2192. provider *SCloudprovider,
  2193. driver cloudprovider.ICloudProvider,
  2194. localRegion *SCloudregion,
  2195. remoteRegion cloudprovider.ICloudRegion,
  2196. syncRange *SSyncRange,
  2197. ) error {
  2198. if syncRange != nil && len(syncRange.Region) > 0 && !utils.IsInStringArray(localRegion.Id, syncRange.Region) {
  2199. // no need to sync
  2200. return nil
  2201. }
  2202. log.Debugf("Start sync cloud provider %s(%s) on region %s(%s)",
  2203. provider.Name, provider.Provider, remoteRegion.GetName(), remoteRegion.GetId())
  2204. storageCachePairs := make([]sStoragecacheSyncPair, 0)
  2205. if cloudprovider.IsSupportQuota(driver) && syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_QUOTA) {
  2206. syncRegionQuotas(ctx, userCred, syncResults, driver, provider, localRegion, remoteRegion, syncRange.Xor)
  2207. }
  2208. localZones, remoteZones, _ := syncRegionZones(ctx, userCred, syncResults, provider, localRegion, remoteRegion, syncRange.Xor)
  2209. if !driver.GetFactory().NeedSyncSkuFromCloud() {
  2210. syncRegionSkus(ctx, userCred, localRegion, syncRange.Xor)
  2211. if syncRange.IsNotSkipSyncResource(NatSkuManager) {
  2212. SyncRegionNatSkus(ctx, userCred, localRegion.Id, true, syncRange.Xor)
  2213. }
  2214. if syncRange.IsNotSkipSyncResource(NasSkuManager) {
  2215. SyncRegionNasSkus(ctx, userCred, localRegion.Id, true, syncRange.Xor)
  2216. }
  2217. } else {
  2218. syncSkusFromPrivateCloud(ctx, userCred, syncResults, localRegion, remoteRegion, syncRange.Xor)
  2219. if cloudprovider.IsSupportRds(driver) && syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_RDS) {
  2220. if syncRange.IsNotSkipSyncResource(DBInstanceManager) {
  2221. syncDBInstanceSkus(ctx, userCred, syncResults, provider, localRegion, remoteRegion, syncRange)
  2222. }
  2223. }
  2224. if cloudprovider.IsSupportNAT(driver) && syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_NAT) {
  2225. if syncRange.IsNotSkipSyncResource(NatSkuManager) {
  2226. syncNATSkus(ctx, userCred, syncResults, provider, localRegion, remoteRegion, syncRange)
  2227. }
  2228. }
  2229. if cloudprovider.IsSupportElasticCache(driver) && syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_CACHE) {
  2230. if syncRange.IsNotSkipSyncResource(ElasticcacheManager) {
  2231. syncCacheSkus(ctx, userCred, syncResults, provider, localRegion, remoteRegion, syncRange)
  2232. }
  2233. }
  2234. }
  2235. // no need to lock public cloud region as cloud region for public cloud is readonly
  2236. if cloudprovider.IsSupportObjectstore(driver) && syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_OBJECTSTORE) {
  2237. if syncRange.IsNotSkipSyncResource(BucketManager) {
  2238. syncRegionBuckets(ctx, userCred, syncResults, provider, localRegion, remoteRegion, syncRange.Xor)
  2239. }
  2240. }
  2241. if cloudprovider.IsSupportCompute(driver) {
  2242. if syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_NETWORK) ||
  2243. syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_NAT) ||
  2244. syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_IMAGE) ||
  2245. syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_EIP) {
  2246. // 需要先同步vpc,避免私有云eip找不到network
  2247. if !(driver.GetFactory().IsPublicCloud() && !syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_NETWORK)) && syncRange.IsNotSkipSyncResource(VpcManager) {
  2248. syncRegionVPCs(ctx, userCred, syncResults, provider, localRegion, remoteRegion, syncRange)
  2249. }
  2250. if syncRange.IsNotSkipSyncResource(ElasticipManager) && syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_EIP) {
  2251. syncRegionEips(ctx, userCred, syncResults, provider, localRegion, remoteRegion, syncRange)
  2252. }
  2253. if syncRange.IsNotSkipSyncResource(SecurityGroupManager) {
  2254. syncRegionSecGroup(ctx, userCred, syncResults, provider, localRegion, remoteRegion, syncRange)
  2255. }
  2256. }
  2257. if syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_COMPUTE) || syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_IMAGE) {
  2258. for j := 0; j < len(localZones); j += 1 {
  2259. if len(syncRange.Zone) > 0 && !utils.IsInStringArray(localZones[j].Id, syncRange.Zone) {
  2260. continue
  2261. }
  2262. // no need to lock zone as public cloud zone is read-only
  2263. newPairs := syncZoneStorages(ctx, userCred, syncResults, provider, driver, &localZones[j], remoteZones[j], syncRange, storageCachePairs)
  2264. if len(newPairs) > 0 {
  2265. storageCachePairs = append(storageCachePairs, newPairs...)
  2266. }
  2267. newPairs = syncZoneHosts(ctx, userCred, syncResults, provider, driver, &localZones[j], remoteZones[j], syncRange, storageCachePairs)
  2268. if len(newPairs) > 0 {
  2269. storageCachePairs = append(storageCachePairs, newPairs...)
  2270. }
  2271. }
  2272. if syncRange.IsNotSkipSyncResource(SnapshotPolicyManager) {
  2273. // sync snapshot policies after sync disks
  2274. syncRegionSnapshotPolicies(ctx, userCred, syncResults, provider, localRegion, remoteRegion, syncRange)
  2275. }
  2276. // sync snapshots after sync disks
  2277. if syncRange.IsNotSkipSyncResource(SnapshotManager) {
  2278. syncRegionSnapshots(ctx, userCred, syncResults, provider, localRegion, remoteRegion, syncRange)
  2279. }
  2280. }
  2281. }
  2282. if cloudprovider.IsSupportNAS(driver) && syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_NAS) {
  2283. if syncRange.IsNotSkipSyncResource(AccessGroupManager) {
  2284. syncRegionAccessGroups(ctx, userCred, syncResults, provider, localRegion, remoteRegion, syncRange)
  2285. }
  2286. if syncRange.IsNotSkipSyncResource(FileSystemManager) {
  2287. syncRegionFileSystems(ctx, userCred, syncResults, provider, localRegion, remoteRegion, syncRange)
  2288. }
  2289. }
  2290. if cloudprovider.IsSupportLoadbalancer(driver) && syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_LOADBALANCER) {
  2291. if syncRange.IsNotSkipSyncResource(LoadbalancerAclManager) {
  2292. syncRegionLoadbalancerAcls(ctx, userCred, syncResults, provider, localRegion, remoteRegion, syncRange)
  2293. }
  2294. if syncRange.IsNotSkipSyncResource(LoadbalancerCertificateManager) {
  2295. syncRegionLoadbalancerCertificates(ctx, userCred, syncResults, provider, localRegion, remoteRegion, syncRange)
  2296. }
  2297. if syncRange.IsNotSkipSyncResource(LoadbalancerHealthCheckManager) {
  2298. syncRegionLoadbalancerHealthChecks(ctx, userCred, syncResults, provider, localRegion, remoteRegion, syncRange)
  2299. }
  2300. if syncRange.IsNotSkipSyncResource(LoadbalancerManager) {
  2301. syncRegionLoadbalancers(ctx, userCred, syncResults, provider, localRegion, remoteRegion, syncRange)
  2302. }
  2303. }
  2304. if cloudprovider.IsSupportCompute(driver) && syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_COMPUTE) {
  2305. if syncRange.IsNotSkipSyncResource(NetworkInterfaceManager) {
  2306. syncRegionNetworkInterfaces(ctx, userCred, syncResults, provider, localRegion, remoteRegion, syncRange)
  2307. }
  2308. }
  2309. if cloudprovider.IsSupportRds(driver) && syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_RDS) {
  2310. if syncRange.IsNotSkipSyncResource(DBInstanceManager) {
  2311. syncRegionDBInstances(ctx, userCred, syncResults, provider, localRegion, remoteRegion, syncRange)
  2312. }
  2313. if syncRange.IsNotSkipSyncResource(DBInstanceBackupManager) {
  2314. syncRegionDBInstanceBackups(ctx, userCred, syncResults, provider, localRegion, remoteRegion, syncRange)
  2315. }
  2316. }
  2317. if cloudprovider.IsSupportElasticCache(driver) && syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_CACHE) {
  2318. if syncRange.IsNotSkipSyncResource(ElasticcacheManager) {
  2319. syncElasticcaches(ctx, userCred, syncResults, provider, localRegion, remoteRegion, syncRange)
  2320. }
  2321. }
  2322. if cloudprovider.IsSupportWaf(driver) && syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_WAF) {
  2323. if syncRange.IsNotSkipSyncResource(WafIPSetManager) {
  2324. syncWafIPSets(ctx, userCred, syncResults, provider, localRegion, remoteRegion, syncRange.Xor)
  2325. }
  2326. if syncRange.IsNotSkipSyncResource(WafRegexSetManager) {
  2327. syncWafRegexSets(ctx, userCred, syncResults, provider, localRegion, remoteRegion, syncRange.Xor)
  2328. }
  2329. if syncRange.IsNotSkipSyncResource(WafInstanceManager) {
  2330. syncWafInstances(ctx, userCred, syncResults, provider, localRegion, remoteRegion, syncRange.Xor)
  2331. }
  2332. }
  2333. if cloudprovider.IsSupportMongoDB(driver) && syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_MONGO_DB) {
  2334. if syncRange.IsNotSkipSyncResource(MongoDBManager) {
  2335. syncMongoDBs(ctx, userCred, syncResults, provider, localRegion, remoteRegion, syncRange.Xor)
  2336. }
  2337. }
  2338. if cloudprovider.IsSupportElasticSearch(driver) && syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_ES) {
  2339. if syncRange.IsNotSkipSyncResource(ElasticSearchManager) {
  2340. syncElasticSearchs(ctx, userCred, syncResults, provider, localRegion, remoteRegion, syncRange.Xor)
  2341. }
  2342. }
  2343. if cloudprovider.IsSupportKafka(driver) && syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_KAFKA) {
  2344. if syncRange.IsNotSkipSyncResource(KafkaManager) {
  2345. syncKafkas(ctx, userCred, syncResults, provider, localRegion, remoteRegion, syncRange.Xor)
  2346. }
  2347. }
  2348. if cloudprovider.IsSupportApp(driver) && syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_APP) {
  2349. if syncRange.IsNotSkipSyncResource(AppManager) {
  2350. syncApps(ctx, userCred, syncResults, provider, localRegion, remoteRegion, syncRange.Xor)
  2351. }
  2352. }
  2353. if cloudprovider.IsSupportContainer(driver) && syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_CONTAINER) {
  2354. if syncRange.IsNotSkipSyncResource(KubeClusterManager) {
  2355. syncKubeClusters(ctx, userCred, syncResults, provider, localRegion, remoteRegion, syncRange.Xor)
  2356. }
  2357. }
  2358. if cloudprovider.IsSupportTablestore(driver) && syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_TABLESTORE) {
  2359. if syncRange.IsNotSkipSyncResource(TablestoreManager) {
  2360. syncTablestore(ctx, userCred, syncResults, provider, localRegion, remoteRegion, syncRange.Xor)
  2361. }
  2362. }
  2363. if cloudprovider.IsSupportModelartsPool(driver) && syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_MODELARTES) {
  2364. if syncRange.IsNotSkipSyncResource(ModelartsPoolManager) {
  2365. syncModelartsPoolSkus(ctx, userCred, syncResults, provider, localRegion, remoteRegion, syncRange.Xor)
  2366. syncModelartsPools(ctx, userCred, syncResults, provider, localRegion, remoteRegion, syncRange.Xor)
  2367. }
  2368. }
  2369. if cloudprovider.IsSupportMiscResources(driver) && syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_MISC) {
  2370. if syncRange.IsNotSkipSyncResource(MiscResourceManager) {
  2371. syncMiscResources(ctx, userCred, syncResults, provider, localRegion, remoteRegion, syncRange.Xor)
  2372. }
  2373. }
  2374. if cloudprovider.IsSupportCompute(driver) && (syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_COMPUTE) || syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_IMAGE)) {
  2375. log.Debugf("storageCachePairs count %d", len(storageCachePairs))
  2376. for i := range storageCachePairs {
  2377. // always sync private cloud cached images
  2378. if storageCachePairs[i].isNew || syncRange.DeepSync || !driver.GetFactory().IsPublicCloud() {
  2379. result := func() compare.SyncResult {
  2380. defer syncResults.AddRequestCost(CachedimageManager)()
  2381. return storageCachePairs[i].syncCloudImages(ctx, userCred, syncRange.Xor)
  2382. }()
  2383. syncResults.Add(CachedimageManager, result)
  2384. notes := fmt.Sprintf("syncCloudImages for %s provider %s result: %s", storageCachePairs[i].local.GetName(), provider.Name, result.Result())
  2385. log.Infof("%s", notes)
  2386. provider.SyncError(result, notes, userCred)
  2387. }
  2388. }
  2389. }
  2390. return nil
  2391. }
  2392. func getZoneForOnPremiseCloudRegion(ctx context.Context, userCred mcclient.TokenCredential, iregion cloudprovider.ICloudRegion) (*SZone, error) {
  2393. extHosts, err := iregion.GetIHosts()
  2394. if err != nil {
  2395. return nil, errors.Wrap(err, "unable to GetIHosts")
  2396. }
  2397. ips := []string{}
  2398. for _, extHost := range extHosts {
  2399. // onpremise host
  2400. accessIp := extHost.GetAccessIp()
  2401. if len(accessIp) == 0 {
  2402. msg := fmt.Sprintf("fail to find wire for host %s: empty host access ip", extHost.GetName())
  2403. log.Errorf("%s", msg)
  2404. continue
  2405. }
  2406. ips = append(ips, accessIp)
  2407. wire, err := WireManager.GetOnPremiseWireOfIp(accessIp)
  2408. if err != nil {
  2409. msg := fmt.Sprintf("fail to find wire for host %s %s: %s", extHost.GetName(), accessIp, err)
  2410. log.Errorf("%s", msg)
  2411. continue
  2412. }
  2413. return wire.GetZone()
  2414. }
  2415. return nil, errors.Wrapf(errors.ErrNotFound, "no suitable zone with accessIp %s", ips)
  2416. }
  2417. func syncOnPremiseCloudProviderStorage(ctx context.Context, userCred mcclient.TokenCredential, syncResults SSyncResultSet, provider *SCloudprovider, iregion cloudprovider.ICloudRegion, driver cloudprovider.ICloudProvider, zone *SZone, syncRange *SSyncRange) []sStoragecacheSyncPair {
  2418. istorages, err := iregion.GetIStorages()
  2419. if err != nil {
  2420. msg := fmt.Sprintf("GetIStorages for zone %s provider %s failed %s", zone.Name, provider.Name, err)
  2421. log.Errorf("%s", msg)
  2422. return nil
  2423. }
  2424. localStorages, remoteStorages, result := StorageManager.SyncStorages(ctx, userCred, provider, zone, istorages, syncRange.Xor)
  2425. syncResults.Add(StorageManager, result)
  2426. msg := result.Result()
  2427. notes := fmt.Sprintf("SyncStorages for zone %s provider %s result: %s", zone.Name, provider.Name, msg)
  2428. log.Infof("%s", notes)
  2429. provider.SyncError(result, notes, userCred)
  2430. if result.IsError() {
  2431. return nil
  2432. }
  2433. storageCachePairs := make([]sStoragecacheSyncPair, 0)
  2434. for i := 0; i < len(localStorages); i += 1 {
  2435. func() {
  2436. lockman.LockObject(ctx, &localStorages[i])
  2437. defer lockman.ReleaseObject(ctx, &localStorages[i])
  2438. if localStorages[i].Deleted {
  2439. return
  2440. }
  2441. if !isInCache(storageCachePairs, localStorages[i].StoragecacheId) {
  2442. cachePair, err := syncStorageCaches(ctx, userCred, provider, &localStorages[i], remoteStorages[i], syncRange.Xor)
  2443. if err != nil {
  2444. log.Errorf("syncStorageCaches for storage %s(%s) error: %v", localStorages[i].Name, localStorages[i].Id, err)
  2445. }
  2446. if cachePair.isValid() {
  2447. storageCachePairs = append(storageCachePairs, cachePair)
  2448. }
  2449. }
  2450. if cloudprovider.IsSupportCompute(driver) && syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_COMPUTE) {
  2451. if !remoteStorages[i].DisableSync() {
  2452. syncStorageDisks(ctx, userCred, syncResults, provider, driver, &localStorages[i], remoteStorages[i], syncRange)
  2453. }
  2454. }
  2455. }()
  2456. }
  2457. return storageCachePairs
  2458. }
  2459. func syncOnPremiseCloudProviderInfo(
  2460. ctx context.Context,
  2461. userCred mcclient.TokenCredential,
  2462. syncResults SSyncResultSet,
  2463. provider *SCloudprovider,
  2464. driver cloudprovider.ICloudProvider,
  2465. syncRange *SSyncRange,
  2466. ) error {
  2467. log.Debugf("Start sync on-premise provider %s(%s)", provider.Name, provider.Provider)
  2468. iregion, err := driver.GetOnPremiseIRegion()
  2469. if err != nil {
  2470. msg := fmt.Sprintf("GetOnPremiseIRegion for provider %s failed %s", provider.GetName(), err)
  2471. log.Errorf("%s", msg)
  2472. return err
  2473. }
  2474. localRegion := CloudregionManager.FetchDefaultRegion()
  2475. if cloudprovider.IsSupportObjectstore(driver) && syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_OBJECTSTORE) {
  2476. if syncRange.IsNotSkipSyncResource(BucketManager) {
  2477. syncRegionBuckets(ctx, userCred, syncResults, provider, localRegion, iregion, syncRange.Xor)
  2478. }
  2479. }
  2480. if cloudprovider.IsSupportNAS(driver) && syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_NAS) {
  2481. if syncRange.IsNotSkipSyncResource(FileSystemManager) {
  2482. syncRegionFileSystems(ctx, userCred, syncResults, provider, localRegion, iregion, syncRange)
  2483. }
  2484. }
  2485. var storageCachePairs []sStoragecacheSyncPair
  2486. if cloudprovider.IsSupportCompute(driver) &&
  2487. (syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_COMPUTE) ||
  2488. syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_IMAGE) ||
  2489. syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_NETWORK)) {
  2490. zone, err := getZoneForOnPremiseCloudRegion(ctx, userCred, iregion)
  2491. if err != nil {
  2492. msg := fmt.Sprintf("Can't get zone for Premise cloud region %s error: %v", iregion.GetName(), err)
  2493. log.Errorf("%s", msg)
  2494. return errors.Wrap(err, "getZoneForOnPremiseCloudRegion")
  2495. }
  2496. if syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_COMPUTE) || syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_NETWORK) {
  2497. remoteVpcs, err := iregion.GetIVpcs()
  2498. if err != nil {
  2499. msg := fmt.Sprintf("GetIVpcs for provider %s failed %s", provider.GetName(), err)
  2500. log.Errorf("%s", msg)
  2501. return err
  2502. }
  2503. {
  2504. // sync wires
  2505. localVpc := VpcManager.FetchDefaultVpc()
  2506. syncVpcWires(ctx, userCred, syncResults, provider, localVpc, remoteVpcs[0], zone, syncRange)
  2507. }
  2508. }
  2509. storageCachePairs = syncOnPremiseCloudProviderStorage(ctx, userCred, syncResults, provider, iregion, driver, zone, syncRange)
  2510. if syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_COMPUTE) {
  2511. ihosts, err := func() ([]cloudprovider.ICloudHost, error) {
  2512. defer syncResults.AddRequestCost(HostManager)()
  2513. return iregion.GetIHosts()
  2514. }()
  2515. if err != nil {
  2516. msg := fmt.Sprintf("GetIHosts for provider %s failed %s", provider.GetName(), err)
  2517. log.Errorf("%s", msg)
  2518. return err
  2519. }
  2520. localHosts, remoteHosts, result := func() ([]SHost, []cloudprovider.ICloudHost, compare.SyncResult) {
  2521. defer syncResults.AddSqlCost(HostManager)()
  2522. return HostManager.SyncHosts(ctx, userCred, provider, zone, nil, ihosts, syncRange.Xor)
  2523. }()
  2524. syncResults.Add(HostManager, result)
  2525. notes := fmt.Sprintf("SyncHosts for zone %s provider %s result: %s", zone.Name, provider.Name, result.Result())
  2526. log.Infof("%s", notes)
  2527. provider.SyncError(result, notes, userCred)
  2528. for i := 0; i < len(localHosts); i += 1 {
  2529. if len(syncRange.Host) > 0 && !utils.IsInStringArray(localHosts[i].Id, syncRange.Host) {
  2530. continue
  2531. }
  2532. newCachePairs := syncHostStorages(ctx, userCred, syncResults, provider, &localHosts[i], remoteHosts[i], storageCachePairs, syncRange.Xor)
  2533. if len(newCachePairs) > 0 {
  2534. storageCachePairs = append(storageCachePairs, newCachePairs...)
  2535. }
  2536. syncHostNics(ctx, userCred, syncResults, provider, &localHosts[i], remoteHosts[i])
  2537. syncHostVMs(ctx, userCred, syncResults, provider, driver, &localHosts[i], remoteHosts[i], syncRange)
  2538. }
  2539. }
  2540. }
  2541. if cloudprovider.IsSupportCompute(driver) && (syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_COMPUTE) || syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_IMAGE)) {
  2542. log.Debugf("storageCachePairs count %d", len(storageCachePairs))
  2543. for i := range storageCachePairs {
  2544. // alway sync on-premise cached images
  2545. // if storageCachePairs[i].isNew || syncRange.DeepSync {
  2546. result := func() compare.SyncResult {
  2547. defer syncResults.AddRequestCost(CachedimageManager)()
  2548. return storageCachePairs[i].syncCloudImages(ctx, userCred, syncRange.Xor)
  2549. }()
  2550. syncResults.Add(CachedimageManager, result)
  2551. notes := fmt.Sprintf("syncCloudImages for stroagecache %s provider %s result: %s", storageCachePairs[i].local.GetId(), provider.Name, result.Result())
  2552. log.Infof("%s", notes)
  2553. provider.SyncError(result, notes, userCred)
  2554. // }
  2555. }
  2556. }
  2557. return nil
  2558. }
  2559. func syncHostNics(ctx context.Context, userCred mcclient.TokenCredential, syncResults SSyncResultSet, provider *SCloudprovider, localHost *SHost, remoteHost cloudprovider.ICloudHost) {
  2560. defer func() {
  2561. if syncResults != nil {
  2562. syncResults.AddSqlCost(NetInterfaceManager)()
  2563. }
  2564. }()
  2565. result := localHost.SyncHostExternalNics(ctx, userCred, remoteHost, provider)
  2566. if syncResults != nil {
  2567. syncResults.Add(NetInterfaceManager, result)
  2568. }
  2569. notes := fmt.Sprintf("SyncHostExternalNics for host %s provider %s result: %s", localHost.Name, provider.Name, result.Result())
  2570. log.Infof("%s", notes)
  2571. }
  2572. func (manager *SCloudproviderregionManager) fetchRecordsByQuery(q *sqlchemy.SQuery) []SCloudproviderregion {
  2573. recs := make([]SCloudproviderregion, 0)
  2574. err := db.FetchModelObjects(manager, q, &recs)
  2575. if err != nil {
  2576. return nil
  2577. }
  2578. return recs
  2579. }
  2580. func (manager *SCloudproviderregionManager) initAllRecords() {
  2581. recs := manager.fetchRecordsByQuery(manager.Query())
  2582. for i := range recs {
  2583. db.Update(&recs[i], func() error {
  2584. recs[i].SyncStatus = api.CLOUD_PROVIDER_SYNC_STATUS_IDLE
  2585. return nil
  2586. })
  2587. }
  2588. }
  2589. func SyncCloudProject(ctx context.Context, userCred mcclient.TokenCredential, model db.IVirtualModel, syncOwnerId mcclient.IIdentityProvider, extModel cloudprovider.IVirtualResource, manager *SCloudprovider) {
  2590. account, err := manager.GetCloudaccount()
  2591. if err != nil {
  2592. return
  2593. }
  2594. projectSync := false
  2595. newOwnerId, err := func() (mcclient.IIdentityProvider, error) {
  2596. rm, err := manager.GetProjectMapping()
  2597. if err != nil {
  2598. if errors.Cause(err) == cloudprovider.ErrNotFound {
  2599. return nil, nil
  2600. }
  2601. return nil, errors.Wrapf(err, "GetProjectMapping")
  2602. }
  2603. if rm != nil && rm.Enabled.Bool() {
  2604. projectSync = rm.IsNeedProjectSync()
  2605. if rm.IsNeedResourceSync() {
  2606. model.SetProjectSrc(apis.OWNER_SOURCE_CLOUD)
  2607. extTags, err := extModel.GetTags()
  2608. if err != nil {
  2609. return nil, errors.Wrapf(err, "extModel.GetTags")
  2610. }
  2611. if rm.Rules != nil {
  2612. for _, rule := range *rm.Rules {
  2613. domainId, projectId, newProj, isMatch := rule.IsMatchTags(extTags)
  2614. if isMatch {
  2615. if len(newProj) > 0 {
  2616. domainId, projectId, err = account.getOrCreateTenant(ctx, newProj, "", "", "auto create from tag", nil)
  2617. if err != nil {
  2618. return nil, errors.Wrapf(err, "getOrCreateTenant(%s)", newProj)
  2619. }
  2620. }
  2621. if len(domainId) > 0 && len(projectId) > 0 {
  2622. return &db.SOwnerId{DomainId: domainId, ProjectId: projectId}, nil
  2623. }
  2624. }
  2625. }
  2626. }
  2627. }
  2628. }
  2629. return nil, nil
  2630. }()
  2631. if err != nil {
  2632. log.Errorf("try sync project for %s %s by tags error: %v", model.Keyword(), model.GetName(), err)
  2633. }
  2634. // 根据云上项目映射或开启同步策略并影响范围为项目标签, 则根据云上项目映射做资源归属
  2635. if extProjectId := extModel.GetProjectId(); len(extProjectId) > 0 && (account.AutoCreateProject || projectSync) && newOwnerId == nil {
  2636. extProject, err := ExternalProjectManager.GetProject(extProjectId, manager.Id)
  2637. if err != nil {
  2638. log.Errorf("sync project for %s %s error: %v", model.Keyword(), model.GetName(), err)
  2639. } else if len(extProject.ProjectId) > 0 {
  2640. newOwnerId = extProject.GetOwnerId()
  2641. }
  2642. }
  2643. if newOwnerId == nil && syncOwnerId != nil && len(syncOwnerId.GetProjectId()) > 0 {
  2644. newOwnerId = syncOwnerId
  2645. }
  2646. if newOwnerId == nil {
  2647. newOwnerId = manager.GetOwnerId()
  2648. }
  2649. model.SyncCloudProjectId(userCred, newOwnerId)
  2650. }
  2651. func SyncCloudDomain(userCred mcclient.TokenCredential, model db.IDomainLevelModel, syncOwnerId mcclient.IIdentityProvider) {
  2652. var newOwnerId mcclient.IIdentityProvider
  2653. if syncOwnerId != nil && len(syncOwnerId.GetProjectDomainId()) > 0 {
  2654. newOwnerId = syncOwnerId
  2655. }
  2656. if newOwnerId == nil {
  2657. newOwnerId = userCred
  2658. }
  2659. model.SyncCloudDomainId(userCred, newOwnerId)
  2660. }
  2661. func syncProjects(ctx context.Context, userCred mcclient.TokenCredential, syncResults SSyncResultSet, cp *SCloudprovider, provider cloudprovider.ICloudProvider, xor bool) error {
  2662. lockman.LockRawObject(ctx, ExternalProjectManager.Keyword(), cp.Id)
  2663. defer lockman.ReleaseRawObject(ctx, ExternalProjectManager.Keyword(), cp.Id)
  2664. projects, err := func() ([]cloudprovider.ICloudProject, error) {
  2665. defer syncResults.AddRequestCost(ExternalProjectManager)()
  2666. return provider.GetIProjects()
  2667. }()
  2668. if err != nil {
  2669. return errors.Wrapf(err, "GetIProjects")
  2670. }
  2671. result := func() compare.SyncResult {
  2672. defer syncResults.AddSqlCost(ExternalProjectManager)()
  2673. return cp.SyncProjects(ctx, userCred, projects, xor)
  2674. }()
  2675. syncResults.Add(ExternalProjectManager, result)
  2676. notes := fmt.Sprintf("SyncProjects for manager %s result: %s", cp.Name, result.Result())
  2677. log.Infof("%s", notes)
  2678. cp.SyncError(result, notes, userCred)
  2679. if result.IsError() {
  2680. return err
  2681. }
  2682. return nil
  2683. }
  2684. func SyncCloudproviderResources(ctx context.Context, userCred mcclient.TokenCredential, provider *SCloudprovider, syncRange *SSyncRange) error {
  2685. driver, err := provider.GetProvider(ctx)
  2686. if err != nil {
  2687. return errors.Wrapf(err, "GetProvider")
  2688. }
  2689. if cloudprovider.IsSupportProject(driver) && syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_PROJECT) {
  2690. err = syncProjects(ctx, userCred, SSyncResultSet{}, provider, driver, syncRange.Xor)
  2691. if err != nil {
  2692. log.Errorf("Sync project for manager %s error: %v", provider.Name, err)
  2693. }
  2694. }
  2695. if cloudprovider.IsSupportCDN(driver) && syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_CDN) {
  2696. err = syncCdnDomains(ctx, userCred, SSyncResultSet{}, provider, driver, syncRange.Xor)
  2697. if err != nil {
  2698. log.Errorf("syncCdnDomains error: %v", err)
  2699. }
  2700. }
  2701. if cloudprovider.IsSupportInterVpcNetwork(driver) && syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_INTERVPCNETWORK) {
  2702. err = syncInterVpcNetworks(ctx, userCred, SSyncResultSet{}, provider, driver, syncRange.Xor)
  2703. if err != nil {
  2704. log.Errorf("syncInterVpcNetworks error: %v", err)
  2705. }
  2706. }
  2707. if cloudprovider.IsSupportDnsZone(driver) && syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_DNSZONE) {
  2708. err = syncDnsZones(ctx, userCred, SSyncResultSet{}, provider, driver, syncRange.Xor)
  2709. if err != nil {
  2710. log.Errorf("syncDnsZones error: %v", err)
  2711. }
  2712. }
  2713. if cloudprovider.IsSupportAiGateway(driver) && syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_AI_GATEWAY) {
  2714. err = syncAiGateways(ctx, userCred, SSyncResultSet{}, provider, driver, syncRange.Xor)
  2715. if err != nil {
  2716. log.Errorf("syncAiGateways error: %v", err)
  2717. }
  2718. }
  2719. if syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_NETWORK) {
  2720. err = syncGlobalVpcs(ctx, userCred, SSyncResultSet{}, provider, driver, syncRange.Xor)
  2721. if err != nil {
  2722. log.Errorf("syncGlobalVpcs error: %v", err)
  2723. }
  2724. }
  2725. if cloudprovider.IsSupportSSLCertificate(driver) && syncRange.NeedSyncResource(cloudprovider.CLOUD_CAPABILITY_CERT) {
  2726. syncSSLCertificates(ctx, userCred, SSyncResultSet{}, provider, driver, syncRange.Xor)
  2727. }
  2728. return nil
  2729. }
  2730. func syncCdnDomains(ctx context.Context, userCred mcclient.TokenCredential, syncResults SSyncResultSet, provider *SCloudprovider, driver cloudprovider.ICloudProvider, xor bool) error {
  2731. domains, err := driver.GetICloudCDNDomains()
  2732. if err != nil {
  2733. return err
  2734. }
  2735. result := provider.SyncCDNDomains(ctx, userCred, domains, xor)
  2736. notes := fmt.Sprintf("Sync CDN for provider %s result: %s", provider.Name, result.Result())
  2737. log.Infof("%s", notes)
  2738. provider.SyncError(result, notes, userCred)
  2739. return nil
  2740. }
  2741. func syncInterVpcNetworks(ctx context.Context, userCred mcclient.TokenCredential, syncResults SSyncResultSet, provider *SCloudprovider, driver cloudprovider.ICloudProvider, xor bool) error {
  2742. networks, err := driver.GetICloudInterVpcNetworks()
  2743. if err != nil {
  2744. return errors.Wrapf(err, "GetICloudInterVpcNetworks")
  2745. }
  2746. localNetwork, remoteNetwork, result := provider.SyncInterVpcNetwork(ctx, userCred, networks, xor)
  2747. notes := fmt.Sprintf("Sync inter vpc network for provider %s result: %s", provider.Name, result.Result())
  2748. log.Infof("%s", notes)
  2749. provider.SyncError(result, notes, userCred)
  2750. for i := range localNetwork {
  2751. lockman.LockObject(ctx, &localNetwork[i])
  2752. defer lockman.ReleaseObject(ctx, &localNetwork[i])
  2753. if localNetwork[i].Deleted {
  2754. continue
  2755. }
  2756. localNetwork[i].SyncInterVpcNetworkRouteSets(ctx, userCred, remoteNetwork[i], xor)
  2757. }
  2758. return nil
  2759. }
  2760. func syncDnsZones(ctx context.Context, userCred mcclient.TokenCredential, syncResults SSyncResultSet, provider *SCloudprovider, driver cloudprovider.ICloudProvider, xor bool) error {
  2761. dnsZones, err := driver.GetICloudDnsZones()
  2762. if err != nil {
  2763. return errors.Wrapf(err, "GetICloudInterVpcNetworks")
  2764. }
  2765. localZones, remoteZones, result := provider.SyncDnsZones(ctx, userCred, dnsZones, xor)
  2766. notes := fmt.Sprintf("Sync dns zones for provider %s result: %s", provider.Name, result.Result())
  2767. log.Infof("%s", notes)
  2768. provider.SyncError(result, notes, userCred)
  2769. for i := range localZones {
  2770. lockman.LockObject(ctx, &localZones[i])
  2771. defer lockman.ReleaseObject(ctx, &localZones[i])
  2772. if localZones[i].Deleted {
  2773. continue
  2774. }
  2775. localZones[i].SyncRecords(ctx, userCred, remoteZones[i], xor)
  2776. }
  2777. return nil
  2778. }
  2779. func syncAiGateways(ctx context.Context, userCred mcclient.TokenCredential, syncResults SSyncResultSet, provider *SCloudprovider, driver cloudprovider.ICloudProvider, xor bool) error {
  2780. gateways, err := driver.GetIAiGateways()
  2781. if err != nil {
  2782. return errors.Wrapf(err, "GetIAiGateways")
  2783. }
  2784. result := func() compare.SyncResult {
  2785. defer syncResults.AddSqlCost(AiGatewayManager)()
  2786. return provider.SyncAiGateways(ctx, userCred, gateways, xor)
  2787. }()
  2788. syncResults.Add(AiGatewayManager, result)
  2789. notes := fmt.Sprintf("Sync ai gateways for provider %s result: %s", provider.Name, result.Result())
  2790. log.Infof("%s", notes)
  2791. provider.SyncError(result, notes, userCred)
  2792. return nil
  2793. }
  2794. func syncGlobalVpcs(ctx context.Context, userCred mcclient.TokenCredential, syncResults SSyncResultSet, provider *SCloudprovider, driver cloudprovider.ICloudProvider, xor bool) error {
  2795. gvpcs, err := driver.GetICloudGlobalVpcs()
  2796. if err != nil {
  2797. if errors.Cause(err) == cloudprovider.ErrNotImplemented || errors.Cause(err) == cloudprovider.ErrNotSupported {
  2798. return nil
  2799. }
  2800. return err
  2801. }
  2802. localVpcs, remoteVpcs, result := provider.SyncGlobalVpcs(ctx, userCred, gvpcs, xor)
  2803. notes := fmt.Sprintf("Sync global vpcs for provider %s result: %s", provider.Name, result.Result())
  2804. log.Infof("%s", notes)
  2805. provider.SyncError(result, notes, userCred)
  2806. for i := range localVpcs {
  2807. lockman.LockObject(ctx, &localVpcs[i])
  2808. defer lockman.ReleaseObject(ctx, &localVpcs[i])
  2809. if localVpcs[i].Deleted {
  2810. continue
  2811. }
  2812. secgroups, err := remoteVpcs[i].GetISecurityGroups()
  2813. if err != nil {
  2814. if errors.Cause(err) == cloudprovider.ErrNotImplemented || errors.Cause(err) == cloudprovider.ErrNotSupported {
  2815. continue
  2816. }
  2817. log.Errorf("GetISecurityGroup for global vpc %s error: %v", localVpcs[i].Name, err)
  2818. continue
  2819. }
  2820. result := localVpcs[i].SyncSecgroups(ctx, userCred, secgroups, xor)
  2821. notes := fmt.Sprintf("Sync security group for global vpc %s provider %s result: %s", localVpcs[i].Name, provider.Name, result.Result())
  2822. log.Infof("%s", notes)
  2823. }
  2824. return nil
  2825. }
  2826. func syncSSLCertificates(ctx context.Context, userCred mcclient.TokenCredential, syncResults SSyncResultSet, provider *SCloudprovider, driver cloudprovider.ICloudProvider, xor bool) error {
  2827. iEss, err := driver.GetISSLCertificates()
  2828. if err != nil {
  2829. return err
  2830. }
  2831. result := provider.SyncSSLCertificates(ctx, userCred, iEss)
  2832. notes := fmt.Sprintf("SyncSSLCertificates for provider %s result: %s", provider.Name, result.Result())
  2833. log.Infof("%s", notes)
  2834. provider.SyncError(result, notes, userCred)
  2835. return nil
  2836. }
  2837. func syncTablestore(
  2838. ctx context.Context,
  2839. userCred mcclient.TokenCredential,
  2840. syncResults SSyncResultSet,
  2841. provider *SCloudprovider,
  2842. localRegion *SCloudregion,
  2843. remoteRegion cloudprovider.ICloudRegion,
  2844. xor bool,
  2845. ) error {
  2846. iTablestores, err := func() ([]cloudprovider.ICloudTablestore, error) {
  2847. defer syncResults.AddRequestCost(TablestoreManager)()
  2848. return remoteRegion.GetICloudTablestores()
  2849. }()
  2850. if err != nil {
  2851. if errors.Cause(err) == cloudprovider.ErrNotImplemented || errors.Cause(err) == cloudprovider.ErrNotSupported {
  2852. return nil
  2853. }
  2854. msg := fmt.Sprintf("GetICloudTablestores for region %s provider %s failed %s", remoteRegion.GetName(), provider.Name, err)
  2855. log.Errorf("%s", msg)
  2856. return err
  2857. }
  2858. result := func() compare.SyncResult {
  2859. defer syncResults.AddSqlCost(TablestoreManager)()
  2860. return localRegion.SyncTablestores(ctx, userCred, iTablestores, provider, xor)
  2861. }()
  2862. syncResults.Add(TablestoreManager, result)
  2863. msg := result.Result()
  2864. notes := fmt.Sprintf("SyncTablestores for region %s provider %s result: %s", localRegion.Name, provider.Name, msg)
  2865. log.Infof("%s", notes)
  2866. provider.SyncError(result, notes, userCred)
  2867. if result.IsError() {
  2868. return result.AllError()
  2869. }
  2870. return nil
  2871. }
  2872. func syncModelartsPools(
  2873. ctx context.Context,
  2874. userCred mcclient.TokenCredential,
  2875. syncResults SSyncResultSet,
  2876. provider *SCloudprovider,
  2877. localRegion *SCloudregion,
  2878. remoteRegion cloudprovider.ICloudRegion,
  2879. xor bool,
  2880. ) error {
  2881. ipools, err := remoteRegion.GetIModelartsPools()
  2882. if err != nil {
  2883. if errors.Cause(err) == cloudprovider.ErrNotImplemented || errors.Cause(err) == cloudprovider.ErrNotSupported {
  2884. return nil
  2885. }
  2886. msg := fmt.Sprintf("GetIModelartsPools for provider %s failed %s", err, ipools)
  2887. log.Errorf("%s", msg)
  2888. return err
  2889. }
  2890. result := localRegion.SyncModelartsPools(ctx, userCred, provider, ipools, xor)
  2891. notes := fmt.Sprintf("SyncModelartsPools for provider %s result: %s", provider.Name, result.Result())
  2892. log.Infof("%s", notes)
  2893. provider.SyncError(result, notes, userCred)
  2894. return nil
  2895. }
  2896. func syncModelartsPoolSkus(
  2897. ctx context.Context,
  2898. userCred mcclient.TokenCredential,
  2899. syncResults SSyncResultSet,
  2900. provider *SCloudprovider,
  2901. localRegion *SCloudregion,
  2902. remoteRegion cloudprovider.ICloudRegion,
  2903. xor bool,
  2904. ) error {
  2905. ipools, err := remoteRegion.GetIModelartsPoolSku()
  2906. if err != nil {
  2907. if errors.Cause(err) == cloudprovider.ErrNotImplemented || errors.Cause(err) == cloudprovider.ErrNotSupported {
  2908. return nil
  2909. }
  2910. msg := fmt.Sprintf("GetIModelartsPoolSku for provider %s provider %s failed %s", provider.Name, err, ipools)
  2911. log.Errorf("%s", msg)
  2912. return err
  2913. }
  2914. result := localRegion.SyncModelartsPoolSkus(ctx, userCred, provider, ipools, xor)
  2915. notes := fmt.Sprintf("SyncModelartsPoolSkus for provider %s result: %s", provider.Name, result.Result())
  2916. log.Infof("%s", notes)
  2917. provider.SyncError(result, notes, userCred)
  2918. return nil
  2919. }
  2920. func syncMiscResources(
  2921. ctx context.Context,
  2922. userCred mcclient.TokenCredential,
  2923. syncResults SSyncResultSet,
  2924. provider *SCloudprovider,
  2925. localRegion *SCloudregion,
  2926. remoteRegion cloudprovider.ICloudRegion,
  2927. xor bool,
  2928. ) error {
  2929. exts, err := remoteRegion.GetIMiscResources()
  2930. if err != nil {
  2931. if errors.Cause(err) == cloudprovider.ErrNotImplemented || errors.Cause(err) == cloudprovider.ErrNotSupported {
  2932. return nil
  2933. }
  2934. msg := fmt.Sprintf("GetIMiscResources for provider %s failed %v", provider.Name, err)
  2935. log.Errorf("%s", msg)
  2936. return err
  2937. }
  2938. result := localRegion.SyncMiscResources(ctx, userCred, provider, exts, xor)
  2939. notes := fmt.Sprintf("SyncMiscResources for provider %s result: %s", provider.Name, result.Result())
  2940. log.Infof("%s", notes)
  2941. provider.SyncError(result, notes, userCred)
  2942. return nil
  2943. }