Merge pull request #4655 from ioito/hotfix/qx-cloudevent-query-optimized

optimized: 优化公有云日志查询

pkg/cloudevent/models/cloudevents.go

@@ -16,34 +16,27 @@ package models
 
 import (
 	"context"
-	"database/sql"
-	"fmt"
+	"time"
 
 	"yunion.io/x/jsonutils"
 	"yunion.io/x/log"
-	"yunion.io/x/pkg/errors"
 	"yunion.io/x/sqlchemy"
 
 	api "yunion.io/x/onecloud/pkg/apis/cloudevent"
 	"yunion.io/x/onecloud/pkg/cloudcommon/db"
-	"yunion.io/x/onecloud/pkg/cloudevent/options"
 	"yunion.io/x/onecloud/pkg/cloudprovider"
-	"yunion.io/x/onecloud/pkg/httperrors"
 	"yunion.io/x/onecloud/pkg/mcclient"
-	"yunion.io/x/onecloud/pkg/mcclient/auth"
-	"yunion.io/x/onecloud/pkg/mcclient/modulebase"
 )
 
 type SCloudeventManager struct {
-	db.SVirtualResourceBaseManager
+	db.SModelBaseManager
 }
 
 var CloudeventManager *SCloudeventManager
-var mods map[string]modulebase.Manager
 
 func init() {
 	CloudeventManager = &SCloudeventManager{
-		SVirtualResourceBaseManager: db.NewVirtualResourceBaseManager(
+		SModelBaseManager: db.NewModelBaseManager(
 			SCloudevent{},
 			"cloudevents_tbl",
 			"cloudevent",
@@ -54,7 +47,7 @@ func init() {
 }
 
 type SCloudevent struct {
-	db.SVirtualResourceBase
+	db.SModelBase
 
 	Service      string               `width:"64" charset:"utf8" nullable:"true" list:"user"`
 	ResourceType string               `width:"64" charset:"utf8" nullable:"true" list:"user"`
@@ -63,8 +56,11 @@ type SCloudevent struct {
 	Request      jsonutils.JSONObject `charset:"utf8" nullable:"true" list:"user"`
 	Account      string               `width:"64" charset:"utf8" nullable:"true" list:"user"`
 	Success      bool                 `nullable:"false" list:"user"`
+	CreatedAt    time.Time            `nullable:"false" created_at:"true" index:"true" get:"user" list:"user"`
 
 	CloudproviderId string `width:"64" charset:"utf8" nullable:"true" list:"user"`
+	Manager         string `width:"128" charset:"utf8" nullable:"false" index:"true" list:"user"`
+	Provider        string `width:"64" charset:"ascii" nullable:"false" list:"user"`
 }
 
 func (self *SCloudeventManager) AllowCreateItem(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, data jsonutils.JSONObject) bool {
@@ -80,116 +76,20 @@ func (self *SCloudevent) AllowUpdateItem(ctx context.Context, userCred mcclient.
 }
 
 func (manager *SCloudeventManager) ListItemFilter(ctx context.Context, q *sqlchemy.SQuery, userCred mcclient.TokenCredential, input *api.CloudeventListInput) (*sqlchemy.SQuery, error) {
-	q, err := manager.SVirtualResourceBaseManager.ListItemFilter(ctx, q, userCred, input.JSON(input))
+	q, err := manager.SModelBaseManager.ListItemFilter(ctx, q, userCred, input.JSON(input))
 	if err != nil {
 		return nil, err
 	}
-	if len(input.Cloudprovider) > 0 {
-		providerObj, err := CloudproviderManager.FetchByIdOrName(userCred, input.Cloudprovider)
-		if err != nil {
-			if err == sql.ErrNoRows {
-				return nil, httperrors.NewResourceNotFoundError2(CloudproviderManager.Keyword(), input.Cloudprovider)
-			} else {
-				return nil, httperrors.NewGeneralError(err)
-			}
-		}
-		q = q.Equals("cloudprovider_id", providerObj.GetId())
-	}
 
 	if len(input.Providers) > 0 {
-		sq := CloudproviderManager.Query().SubQuery()
-		q = q.Join(sq, sqlchemy.Equals(q.Field("cloudprovider_id"), sq.Field("id"))).
-			Filter(sqlchemy.In(sq.Field("provider"), input.Providers))
+		q = q.In("provider", input.Providers)
 	}
-	//过滤已删除的cloudprovider日志
-	sq := CloudproviderManager.Query("id").SubQuery()
-	q = q.In("cloudprovider_id", sq)
+
 	return q, nil
 }
 
-func (self *SCloudevent) GetCloudprovider() (*SCloudprovider, error) {
-	cloudprovider, err := CloudproviderManager.FetchById(self.CloudproviderId)
-	if err != nil {
-		return nil, err
-	}
-	return cloudprovider.(*SCloudprovider), nil
-}
-
 func (self *SCloudevent) GetCustomizeColumns(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject) *jsonutils.JSONDict {
-	extra := self.SStatusStandaloneResourceBase.GetCustomizeColumns(ctx, userCred, query)
-	extra, _ = self.getMoreDetails(ctx, userCred, query, extra)
-	return extra
-}
-
-func (self *SCloudevent) getMoreDetails(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject, extra *jsonutils.JSONDict) (*jsonutils.JSONDict, error) {
-	cloudprovider, err := self.GetCloudprovider()
-	if err != nil {
-		return nil, err
-	}
-	info := jsonutils.Marshal(map[string]string{
-		"provider": cloudprovider.Provider,
-		"manager":  cloudprovider.Name,
-	})
-	extra.Update(info)
-	return extra, nil
-}
-
-func (self *SCloudevent) GetExtraDetails(ctx context.Context, userCred mcclient.TokenCredential, query jsonutils.JSONObject) (*jsonutils.JSONDict, error) {
-	extra, err := self.SVirtualResourceBase.GetExtraDetails(ctx, userCred, query)
-	if err != nil {
-		return nil, err
-	}
-	return self.getMoreDetails(ctx, userCred, query, extra)
-}
-
-func (self *SCloudeventManager) fetchMods(ctx context.Context, userCred mcclient.TokenCredential) {
-	if len(mods) > 0 {
-		return
-	}
-	s := auth.GetAdminSession(ctx, options.Options.Region, "v2")
-	mods = map[string]modulebase.Manager{}
-	rms, _ := modulebase.GetRegisterdModules()
-	for _, _mods := range rms {
-		for _, _mod := range _mods {
-			if _, ok := mods[_mod]; !ok {
-				mod, err := modulebase.GetModule(s, _mod)
-				if err != nil {
-					log.Errorf("failed to get mod %s error: %v", _mod, err)
-					continue
-				}
-				mods[mod.GetKeyword()] = mod
-			}
-		}
-	}
-	return
-}
-
-func (manager *SCloudeventManager) setEventInfo(session *mcclient.ClientSession, mod modulebase.Manager, event *SCloudevent) error {
-	params := jsonutils.NewDict()
-	params.Add(jsonutils.NewString(fmt.Sprintf("external_id.equals(%s)", event.Name)), "filter")
-	result, err := mod.List(session, params)
-	if err != nil {
-		return errors.Wrapf(err, "mod.List for %s by externalId: %s", mod.KeyString(), event.Name)
-	}
-	if len(result.Data) != 1 {
-		return errors.Wrapf(err, "found %d %s by externalId: %s", len(result.Data), mod.KeyString(), event.Name)
-	}
-
-	data := struct {
-		Name     string
-		TenantId string
-	}{}
-	err = result.Data[0].Unmarshal(&data)
-	if err != nil {
-		return errors.Wrapf(err, "result.Data[0].Unmarshal %s", result.Data[0])
-	}
-	if len(data.Name) > 0 {
-		event.Name = event.Name
-	}
-	if len(data.TenantId) > 0 {
-		event.ProjectId = data.TenantId
-	}
-	return nil
+	return self.SModelBase.GetCustomizeColumns(ctx, userCred, query)
 }
 
 func (manager *SCloudeventManager) SyncCloudevent(ctx context.Context, userCred mcclient.TokenCredential, cloudprovider *SCloudprovider, iEvents []cloudprovider.ICloudEvent) int {
@@ -203,16 +103,12 @@ func (manager *SCloudeventManager) SyncCloudevent(ctx context.Context, userCred
 			RequestId:       iEvent.GetRequestId(),
 			Request:         iEvent.GetRequest(),
 			Success:         iEvent.IsSuccess(),
+			Manager:         cloudprovider.Name,
+			Provider:        cloudprovider.Provider,
 			CloudproviderId: cloudprovider.Id,
 		}
 
-		event.Name = iEvent.GetName()
-		event.Status = "ready"
-		event.ProjectId = userCred.GetProjectId()
 		event.CreatedAt = iEvent.GetCreatedAt()
-		event.ProjectId = userCred.GetProjectId()
-		event.DomainId = userCred.GetDomainId()
-
 		event.SetModelManager(manager, event)
 		err := manager.TableSpec().Insert(event)
 		if err != nil {
@@ -223,3 +119,11 @@ func (manager *SCloudeventManager) SyncCloudevent(ctx context.Context, userCred
 	}
 	return count
 }
+
+func (manager *SCloudeventManager) GetPagingConfig() *db.SPagingConfig {
+	return &db.SPagingConfig{
+		Order:        sqlchemy.SQL_ORDER_DESC,
+		MarkerField:  "created_at",
+		DefaultLimit: 20,
+	}
+}

pkg/cloudevent/models/cloudproviders.go

@@ -56,10 +56,11 @@ func init() {
 type SCloudprovider struct {
 	db.SEnabledStatusStandaloneResourceBase
 
-	HealthStatus  string `width:"16" charset:"ascii" default:"normal" nullable:"false" list:"domain"`
-	SyncStatus    string
-	LastSync      time.Time
-	LastSyncEndAt time.Time
+	HealthStatus   string `width:"16" charset:"ascii" default:"normal" nullable:"false" list:"domain"`
+	SyncStatus     string
+	LastSync       time.Time
+	LastSyncEndAt  time.Time
+	LastSyncTimeAt time.Time
 
 	AccessUrl string `width:"64" charset:"ascii" nullable:"true" list:"domain" update:"domain"`
 	Account   string `width:"128" charset:"ascii" nullable:"false" list:"domain"`
@@ -182,6 +183,18 @@ func (self *SCloudprovider) MarkEndSync(userCred mcclient.TokenCredential) error
 	return nil
 }
 
+func (self *SCloudprovider) SetLastSyncTimeAt(userCred mcclient.TokenCredential, last time.Time) error {
+	_, err := db.Update(self, func() error {
+		self.LastSyncTimeAt = last
+		return nil
+	})
+	if err != nil {
+		log.Errorf("Failed to SetLastSyncTimeAt error: %v", err)
+		return err
+	}
+	return nil
+}
+
 func (manager *SCloudproviderManager) newFromRegionProvider(ctx context.Context, userCred mcclient.TokenCredential, cloudprovider SCloudprovider) error {
 	cloudprovider.SyncStatus = api.CLOUD_PROVIDER_SYNC_STATUS_IDLE
 	return manager.TableSpec().Insert(&cloudprovider)
@@ -226,19 +239,23 @@ func (self *SCloudprovider) GetNextTimeRange() (time.Time, time.Time, error) {
 	if err != nil {
 		return start, end, errors.Wrap(err, "q.CountWithError")
 	}
-	if count == 0 {
+	if !self.LastSyncTimeAt.IsZero() {
+		start = self.LastSyncTimeAt
+	} else if count == 0 {
 		start = time.Now().AddDate(0, 0, -1*factory.GetMaxCloudEventKeepDays())
 	} else {
-		provider := SCloudprovider{}
-		err = q.First(&provider)
+		event := &SCloudevent{}
+		err = q.First(event)
 		if err != nil {
 			return start, end, errors.Wrap(err, "q.First")
 		}
-		start = provider.CreatedAt
-		if start.Before(time.Now().AddDate(0, 0, factory.GetMaxCloudEventKeepDays()*-1)) {
-			start = time.Now().AddDate(0, 0, factory.GetMaxCloudEventKeepDays()*-1)
-		}
+		start = event.CreatedAt
 	}
+	// 避免cloudevent过长时间未运行，再次运行时记录的最后一条时间距离现在间隔太长
+	if start.Before(time.Now().AddDate(0, 0, factory.GetMaxCloudEventKeepDays()*-1)) {
+		start = time.Now().AddDate(0, 0, factory.GetMaxCloudEventKeepDays()*-1)
+	}
+
 	if options.Options.OneSyncForHours > factory.GetMaxCloudEventSyncDays()*24 {
 		end = start.Add(time.Duration(factory.GetMaxCloudEventSyncDays()*24) * time.Hour)
 	} else {

pkg/cloudevent/tasks/cloudevent_sync_task.go

@@ -97,6 +97,7 @@ func (self *CloudeventSyncTask) OnInit(ctx context.Context, obj db.IStandaloneMo
 		log.Infof("Sync %d events for %s(%s) from %s(%d) hours", _count, provider.Name, provider.Id, start.Format("2006-01-02T15:04:05Z"), duration/time.Hour)
 		count += _count
 
+		provider.SetLastSyncTimeAt(self.UserCred, end)
 		if time.Now().Sub(end) < duration {
 			break
 		}

pkg/mcclient/modules/mod_cloudevents.go

@@ -22,8 +22,8 @@ var (
 
 func init() {
 	Cloudevents = NewCloudeventManager("cloudevent", "cloudevents",
-		[]string{"ID", "Name", "Status", "Service", "Success",
-			"Resource_Type", "Action", "Cloudprovider_Id", "Cloudprovider"},
+		[]string{"Action", "Service", "Success",
+			"Resource_Type", "Cloudprovider_Id", "Manager", "Provider"},
 		[]string{})
 
 	register(&Cloudevents)