feat(region): sync resource tag

pkg/apis/compute/elastic_search.go

@@ -29,6 +29,11 @@ const (
 	ELASTIC_SEARCH_STATUS_UNKNOWN       = "unknown"
 )
 
+const (
+	ELASTIC_SEARCH_UPDATE_TAGS        = "update_tags"
+	ELASTIC_SEARCH_UPDATE_TAGS_FAILED = "update_tags_fail"
+)
+
 // 资源创建参数, 目前仅占位
 type ElasticSearchCreateInput struct {
 }

pkg/compute/models/elastic_search.go

@@ -658,3 +658,31 @@ func (self *SElasticSearch) GetDetailsAccessInfo(ctx context.Context, userCred m
 func (es *SElasticSearch) PostUpdate(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data jsonutils.JSONObject) {
 	es.SVirtualResourceBase.PostUpdate(ctx, userCred, query, data)
 }
+
+func (self *SElasticSearch) StartSElasticSearchSyncTask(ctx context.Context, userCred mcclient.TokenCredential, parentTaskId string) error {
+	return StartResourceSyncStatusTask(ctx, userCred, self, "ElasticSearchSyncstatusTask", parentTaskId)
+}
+
+func (self *SElasticSearch) StartRemoteUpdateTask(ctx context.Context, userCred mcclient.TokenCredential, replaceTags bool, parentTaskId string) error {
+	data := jsonutils.NewDict()
+	if replaceTags {
+		data.Add(jsonutils.JSONTrue, "replace_tags")
+	}
+	if task, err := taskman.TaskManager.NewTask(ctx, "ElasticSearchRemoteUpdateTask", self, userCred, data, parentTaskId, "", nil); err != nil {
+		return errors.Wrap(err, "Start ElasticSearchRemoteUpdateTask")
+	} else {
+		self.SetStatus(userCred, api.ELASTIC_SEARCH_UPDATE_TAGS, "StartRemoteUpdateTask")
+		task.ScheduleRun(nil)
+	}
+	return nil
+}
+
+func (self *SElasticSearch) OnMetadataUpdated(ctx context.Context, userCred mcclient.TokenCredential) {
+	if len(self.ExternalId) == 0 {
+		return
+	}
+	err := self.StartRemoteUpdateTask(ctx, userCred, true, "")
+	if err != nil {
+		log.Errorf("StartRemoteUpdateTask fail: %s", err)
+	}
+}

pkg/compute/models/regiondrivers.go

@@ -38,6 +38,7 @@ type IRegionDriver interface {
 	IElasticcacheAcl
 	IElasticcacheBackup
 	IDBInstanceDriver
+	IElasticSearchDriver
 
 	ValidateCreateLoadbalancerData(ctx context.Context, userCred mcclient.TokenCredential, ownerId mcclient.IIdentityProvider, data *api.LoadbalancerCreateInput) (*api.LoadbalancerCreateInput, error)
 	RequestCreateLoadbalancerInstance(ctx context.Context, userCred mcclient.TokenCredential, lb *SLoadbalancer, input *api.LoadbalancerCreateInput, task taskman.ITask) error
@@ -270,6 +271,10 @@ type IElasticIpDriver interface {
 	RequestAssociateEip(ctx context.Context, userCred mcclient.TokenCredential, eip *SElasticip, input api.ElasticipAssociateInput, obj db.IStatusStandaloneModel, task taskman.ITask) error
 }
 
+type IElasticSearchDriver interface {
+	RequestRemoteUpdateElasticSearch(ctx context.Context, userCred mcclient.TokenCredential, elasticcache *SElasticSearch, replaceTags bool, task taskman.ITask) error
+}
+
 var regionDrivers map[string]IRegionDriver
 
 func init() {

pkg/compute/regiondrivers/base.go

@@ -521,3 +521,7 @@ func (self *SBaseRegionDriver) RequestPackInstanceBackup(ctx context.Context, ib
 func (self *SBaseRegionDriver) RequestUnpackInstanceBackup(ctx context.Context, ib *models.SInstanceBackup, task taskman.ITask, packageName string, metadataOnly bool) error {
 	return errors.Wrapf(cloudprovider.ErrNotImplemented, "RequestUnpackInstanceBackup")
 }
+
+func (self *SBaseRegionDriver) RequestRemoteUpdateElasticSearch(ctx context.Context, userCred mcclient.TokenCredential, elasticcache *models.SElasticSearch, replaceTags bool, task taskman.ITask) error {
+	return errors.Wrapf(cloudprovider.ErrNotImplemented, "RequestRemoteUpdateElasticSearch")
+}

pkg/compute/regiondrivers/managedvirtual.go

@@ -3248,3 +3248,35 @@ func (self *SManagedVirtualizationRegionDriver) RequestCreateNetwork(ctx context
 
 	return net.SyncWithCloudNetwork(ctx, userCred, inet, nil, nil)
 }
+
+func (self *SManagedVirtualizationRegionDriver) RequestRemoteUpdateElasticSearch(ctx context.Context, userCred mcclient.TokenCredential, instance *models.SElasticSearch, replaceTags bool, task taskman.ITask) error {
+	taskman.LocalTaskRun(task, func() (jsonutils.JSONObject, error) {
+		ies, err := instance.GetIElasticSearch(ctx)
+		if err != nil {
+			return nil, errors.Wrap(err, "instance.GetIESInstance")
+		}
+		oldTags, err := ies.GetTags()
+		if err != nil {
+			if errors.Cause(err) == cloudprovider.ErrNotSupported || errors.Cause(err) == cloudprovider.ErrNotImplemented {
+				return nil, nil
+			}
+			return nil, errors.Wrap(err, "ies.GetTags()")
+		}
+		tags, err := instance.GetAllUserMetadata()
+		if err != nil {
+			return nil, errors.Wrapf(err, "instance.GetAllUserMetadata")
+		}
+		tagsUpdateInfo := cloudprovider.TagsUpdateInfo{OldTags: oldTags, NewTags: tags}
+		err = cloudprovider.SetTags(ctx, ies, instance.ManagerId, tags, replaceTags)
+		if err != nil {
+			if errors.Cause(err) == cloudprovider.ErrNotSupported || errors.Cause(err) == cloudprovider.ErrNotImplemented {
+				return nil, nil
+			}
+			logclient.AddActionLogWithStartable(task, instance, logclient.ACT_UPDATE_TAGS, err, userCred, false)
+			return nil, errors.Wrap(err, "ies.SetTags")
+		}
+		logclient.AddActionLogWithStartable(task, instance, logclient.ACT_UPDATE_TAGS, tagsUpdateInfo, userCred, true)
+		return nil, nil
+	})
+	return nil
+}

pkg/compute/tasks/elastic_search_remote_update_task.go

@@ -0,0 +1,72 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package tasks
+
+import (
+	"context"
+	"fmt"
+
+	"yunion.io/x/jsonutils"
+
+	api "yunion.io/x/onecloud/pkg/apis/compute"
+	"yunion.io/x/onecloud/pkg/cloudcommon/db"
+	"yunion.io/x/onecloud/pkg/cloudcommon/db/taskman"
+	"yunion.io/x/onecloud/pkg/compute/models"
+)
+
+type ElasticSearchRemoteUpdateTask struct {
+	taskman.STask
+}
+
+func init() {
+	taskman.RegisterTask(ElasticSearchRemoteUpdateTask{})
+}
+
+func (self *ElasticSearchRemoteUpdateTask) taskFail(ctx context.Context, elasticcache *models.SElasticSearch, reason jsonutils.JSONObject) {
+	elasticcache.SetStatus(self.UserCred, api.ELASTIC_SEARCH_UPDATE_TAGS_FAILED, reason.String())
+	self.SetStageFailed(ctx, reason)
+}
+
+func (self *ElasticSearchRemoteUpdateTask) OnInit(ctx context.Context, obj db.IStandaloneModel, data jsonutils.JSONObject) {
+	es := obj.(*models.SElasticSearch)
+	region, _ := es.GetRegion()
+	if region == nil {
+		self.taskFail(ctx, es, jsonutils.NewString(fmt.Sprintf("failed to find region for elastic search %s", es.GetName())))
+		return
+	}
+	self.SetStage("OnRemoteUpdateComplete", nil)
+	replaceTags := jsonutils.QueryBoolean(self.Params, "replace_tags", false)
+
+	if err := region.GetDriver().RequestRemoteUpdateElasticSearch(ctx, self.GetUserCred(), es, replaceTags, self); err != nil {
+		self.taskFail(ctx, es, jsonutils.NewString(err.Error()))
+	}
+}
+
+func (self *ElasticSearchRemoteUpdateTask) OnRemoteUpdateComplete(ctx context.Context, elasticcache *models.SElasticSearch, data jsonutils.JSONObject) {
+	self.SetStage("OnSyncStatusComplete", nil)
+	elasticcache.StartSElasticSearchSyncTask(ctx, self.UserCred, self.GetTaskId())
+}
+
+func (self *ElasticSearchRemoteUpdateTask) OnRemoteUpdateCompleteFailed(ctx context.Context, elasticcache *models.SElasticSearch, data jsonutils.JSONObject) {
+	self.taskFail(ctx, elasticcache, data)
+}
+
+func (self *ElasticSearchRemoteUpdateTask) OnSyncStatusComplete(ctx context.Context, elasticcache *models.SElasticSearch, data jsonutils.JSONObject) {
+	self.SetStageComplete(ctx, nil)
+}
+
+func (self *ElasticSearchRemoteUpdateTask) OnSyncStatusCompleteFailed(ctx context.Context, elasticcache *models.SElasticSearch, data jsonutils.JSONObject) {
+	self.SetStageFailed(ctx, data)
+}