From f03cbec6532cfadface6a55b365d36df879e8069 Mon Sep 17 00:00:00 2001 From: withchao <993506633@qq.com> Date: Wed, 18 Dec 2024 17:41:49 +0800 Subject: [PATCH] refactoring scheduled tasks --- internal/rpc/third/s3.go | 93 +++------------- internal/rpc/third/third.go | 14 +-- internal/tools/cron_task.go | 125 ++++++++-------------- internal/tools/msg.go | 23 ++++ internal/tools/s3.go | 34 ++++++ internal/tools/user_msg.go | 31 ++++++ pkg/common/storage/controller/s3.go | 20 +--- pkg/common/storage/database/mgo/object.go | 29 +++-- pkg/common/storage/database/object.go | 6 +- 9 files changed, 175 insertions(+), 200 deletions(-) create mode 100644 internal/tools/msg.go create mode 100644 internal/tools/s3.go create mode 100644 internal/tools/user_msg.go diff --git a/internal/rpc/third/s3.go b/internal/rpc/third/s3.go index b4b064d7f..d4146fdf4 100644 --- a/internal/rpc/third/s3.go +++ b/internal/rpc/third/s3.go @@ -23,13 +23,9 @@ import ( "strconv" "time" - "github.com/openimsdk/open-im-server/v3/pkg/common/config" - "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model" - "go.mongodb.org/mongo-driver/mongo" - "github.com/google/uuid" "github.com/openimsdk/open-im-server/v3/pkg/common/servererrs" - "github.com/openimsdk/protocol/sdkws" + "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model" "github.com/openimsdk/protocol/third" "github.com/openimsdk/tools/errs" "github.com/openimsdk/tools/log" @@ -288,87 +284,30 @@ func (t *thirdServer) apiAddress(prefix, name string) string { } func (t *thirdServer) DeleteOutdatedData(ctx context.Context, req *third.DeleteOutdatedDataReq) (*third.DeleteOutdatedDataResp, error) { - var conf config.Third + engine := t.config.RpcConfig.Object.Enable expireTime := time.UnixMilli(req.ExpireTime) - - findPagination := &sdkws.RequestPagination{ - PageNumber: 1, - ShowNumber: 500, - } - // Find all expired data in S3 database - total, models, err := t.s3dataBase.FindNeedDeleteObjectByDB(ctx, expireTime, req.ObjectGroup, findPagination) - if err != nil && errs.Unwrap(err) != mongo.ErrNoDocuments { - return nil, errs.Wrap(err) + models, err := t.s3dataBase.FindExpirationObject(ctx, engine, expireTime, req.ObjectGroup, int64(req.Count)) + if err != nil { + return nil, err } - - if total == 0 { - log.ZDebug(ctx, "Not have OutdatedData", "delete Total", total) - return &third.DeleteOutdatedDataResp{Count: int32(total)}, nil - } - - needDelObjectKeys := make([]string, len(models)) - for _, model := range models { - needDelObjectKeys = append(needDelObjectKeys, model.Key) - } - - // Remove duplicate keys, have the same key use in different models - needDelObjectKeys = datautil.Distinct(needDelObjectKeys) - - for _, key := range needDelObjectKeys { - // Find all models by key - keyModels, err := t.s3dataBase.FindModelsByKey(ctx, key) - if err != nil && errs.Unwrap(err) != mongo.ErrNoDocuments { + if len(models) > 0 { + names := datautil.Batch(func(o *model.Object) string { + return o.Name + }, models) + if err := t.s3dataBase.DeleteSpecifiedData(ctx, engine, names); err != nil { return nil, errs.Wrap(err) } - - // check keyModels, if all keyModels. - needDelKey := true // Default can delete - for _, keymodel := range keyModels { - // If group is empty or CreateTime is after expireTime, can't delete this key - if keymodel.Group == "" || keymodel.CreateTime.After(expireTime) { - needDelKey = false - break - } + if err := t.s3dataBase.DelS3Key(ctx, engine, names...); err != nil { + return nil, err } - - // If this object is not referenced by not expire data, delete it - if needDelKey && t.minio != nil { - // If have a thumbnail, delete it - thumbnailKey, _ := t.getMinioImageThumbnailKey(ctx, key) - if thumbnailKey != "" { - err := t.s3dataBase.DeleteObject(ctx, thumbnailKey) - if err != nil { - log.ZWarn(ctx, "Delete thumbnail object is error:", errs.Wrap(err), "thumbnailKey", thumbnailKey) - } - } - - // Delete object - err = t.s3dataBase.DeleteObject(ctx, key) - if err != nil { - log.ZWarn(ctx, "Delete object is error", errs.Wrap(err), "object key", key) - } - - // Delete cache key - err = t.s3dataBase.DelS3Key(ctx, conf.Object.Enable, key) - if err != nil { - log.ZWarn(ctx, "Delete cache key is error:", errs.Wrap(err), "cache S3 key:", key) + for _, object := range models { + if err := t.s3.DeleteObject(ctx, object.Key); err != nil { + return nil, err } } } - - // handle delete data in S3 database - for _, model := range models { - // Delete all expired data row in S3 database - err := t.s3dataBase.DeleteSpecifiedData(ctx, model.Engine, model.Name) - if err != nil { - return nil, errs.Wrap(err) - } - } - - log.ZDebug(ctx, "DeleteOutdatedData", "delete Total", total) - - return &third.DeleteOutdatedDataResp{Count: int32(total)}, nil + return &third.DeleteOutdatedDataResp{Count: int32(len(models))}, nil } type FormDataMate struct { diff --git a/internal/rpc/third/third.go b/internal/rpc/third/third.go index 77e6d459f..dc964fdb1 100644 --- a/internal/rpc/third/third.go +++ b/internal/rpc/third/third.go @@ -44,7 +44,7 @@ type thirdServer struct { s3dataBase controller.S3Database defaultExpire time.Duration config *Config - minio *minio.Minio + s3 s3.Interface } type Config struct { @@ -79,13 +79,11 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg // Select the oss method according to the profile policy enable := config.RpcConfig.Object.Enable var ( - o s3.Interface - minioCli *minio.Minio + o s3.Interface ) switch enable { case "minio": - minioCli, err = minio.NewMinio(ctx, redis.NewMinioCache(rdb), *config.MinioConfig.Build()) - o = minioCli + o, err = minio.NewMinio(ctx, redis.NewMinioCache(rdb), *config.MinioConfig.Build()) case "cos": o, err = cos.NewCos(*config.RpcConfig.Object.Cos.Build()) case "oss": @@ -106,15 +104,11 @@ func Start(ctx context.Context, config *Config, client discovery.SvcDiscoveryReg s3dataBase: controller.NewS3Database(rdb, o, s3db), defaultExpire: time.Hour * 24 * 7, config: config, - minio: minioCli, + s3: o, }) return nil } -func (t *thirdServer) getMinioImageThumbnailKey(ctx context.Context, name string) (string, error) { - return t.minio.GetImageThumbnailKey(ctx, name) -} - func (t *thirdServer) FcmUpdateToken(ctx context.Context, req *third.FcmUpdateTokenReq) (resp *third.FcmUpdateTokenResp, err error) { err = t.thirdDatabase.FcmUpdateToken(ctx, req.Account, int(req.PlatformID), req.FcmToken, req.ExpireTime) if err != nil { diff --git a/internal/tools/cron_task.go b/internal/tools/cron_task.go index 2fe7d0e39..049a6199c 100644 --- a/internal/tools/cron_task.go +++ b/internal/tools/cron_task.go @@ -2,10 +2,6 @@ package tools import ( "context" - "fmt" - "os" - "time" - "github.com/openimsdk/open-im-server/v3/pkg/common/config" kdisc "github.com/openimsdk/open-im-server/v3/pkg/common/discoveryregister" pbconversation "github.com/openimsdk/protocol/conversation" @@ -60,87 +56,58 @@ func Start(ctx context.Context, config *CronTaskConfig) error { return err } - msgClient := msg.NewMsgClient(msgConn) - conversationClient := pbconversation.NewConversationClient(conversationConn) - thirdClient := third.NewThirdClient(thirdConn) - - crontab := cron.New() - - // scheduled hard delete outdated Msgs in specific time. - destructMsgsFunc := func() { - now := time.Now() - deltime := now.Add(-time.Hour * 24 * time.Duration(config.CronTask.RetainChatRecords)) - ctx := mcontext.SetOperationID(ctx, fmt.Sprintf("cron_%d_%d", os.Getpid(), deltime.UnixMilli())) - log.ZDebug(ctx, "Destruct chat records", "deltime", deltime, "timestamp", deltime.UnixMilli()) - - if _, err := msgClient.DestructMsgs(ctx, &msg.DestructMsgsReq{Timestamp: deltime.UnixMilli()}); err != nil { - log.ZError(ctx, "cron destruct chat records failed", err, "deltime", deltime, "cont", time.Since(now)) - return - } - log.ZDebug(ctx, "cron destruct chat records success", "deltime", deltime, "cont", time.Since(now)) - } - if _, err := crontab.AddFunc(config.CronTask.CronExecuteTime, destructMsgsFunc); err != nil { - return errs.Wrap(err) + srv := &cronServer{ + ctx: ctx, + config: config, + cron: cron.New(), + msgClient: msg.NewMsgClient(msgConn), + conversationClient: pbconversation.NewConversationClient(conversationConn), + thirdClient: third.NewThirdClient(thirdConn), } - // scheduled soft delete outdated Msgs in specific time when user set `is_msg_destruct` feature. - clearMsgFunc := func() { - now := time.Now() - ctx := mcontext.SetOperationID(ctx, fmt.Sprintf("cron_%d_%d", os.Getpid(), now.UnixMilli())) - log.ZDebug(ctx, "clear msg cron start", "now", now) - - conversations, err := conversationClient.GetConversationsNeedClearMsg(ctx, &pbconversation.GetConversationsNeedClearMsgReq{}) - if err != nil { - log.ZError(ctx, "Get conversation need Destruct msgs failed.", err) - return - } - - _, err = msgClient.ClearMsg(ctx, &msg.ClearMsgReq{Conversations: conversations.Conversations}) - if err != nil { - log.ZError(ctx, "Clear Msg failed.", err) - return - } - - log.ZDebug(ctx, "clear msg cron task completed", "cont", time.Since(now)) + if err := srv.registerClearS3(); err != nil { + return err } - if _, err := crontab.AddFunc(config.CronTask.CronExecuteTime, clearMsgFunc); err != nil { - return errs.Wrap(err) + if err := srv.registerDeleteMsg(); err != nil { + return err } - - // scheduled delete outdated file Objects and their datas in specific time. - deleteObjectFunc := func() { - now := time.Now() - executeNum := 5 - // number of pagination. if need modify, need update value in third.DeleteOutdatedData - pageShowNumber := 500 - deleteTime := now.Add(-time.Hour * 24 * time.Duration(config.CronTask.FileExpireTime)) - ctx := mcontext.SetOperationID(ctx, fmt.Sprintf("cron_%d_%d", os.Getpid(), deleteTime.UnixMilli())) - log.ZDebug(ctx, "deleteoutDatedData", "deletetime", deleteTime, "timestamp", deleteTime.UnixMilli()) - - if len(config.CronTask.DeleteObjectType) == 0 { - log.ZDebug(ctx, "cron deleteoutDatedData not type need delete", "deletetime", deleteTime, "DeleteObjectType", config.CronTask.DeleteObjectType, "cont", time.Since(now)) - return - } - - for i := 0; i < executeNum; i++ { - resp, err := thirdClient.DeleteOutdatedData(ctx, &third.DeleteOutdatedDataReq{ExpireTime: deleteTime.UnixMilli(), ObjectGroup: config.CronTask.DeleteObjectType}) - if err != nil { - log.ZError(ctx, "cron deleteoutDatedData failed", err, "deleteTime", deleteTime, "cont", time.Since(now)) - return - } - if resp.Count == 0 || resp.Count < int32(pageShowNumber) { - break - } - } - - log.ZDebug(ctx, "cron deleteoutDatedData success", "deltime", deleteTime, "cont", time.Since(now)) + if err := srv.registerClearUserMsg(); err != nil { + return err } - if _, err := crontab.AddFunc(config.CronTask.CronExecuteTime, deleteObjectFunc); err != nil { - return errs.Wrap(err) - } - log.ZDebug(ctx, "start cron task", "CronExecuteTime", config.CronTask.CronExecuteTime) - crontab.Start() + srv.cron.Start() <-ctx.Done() return nil } + +type cronServer struct { + ctx context.Context + config *CronTaskConfig + cron *cron.Cron + msgClient msg.MsgClient + conversationClient pbconversation.ConversationClient + thirdClient third.ThirdClient +} + +func (c *cronServer) registerClearS3() error { + if c.config.CronTask.FileExpireTime <= 0 || len(c.config.CronTask.DeleteObjectType) == 0 { + log.ZInfo(c.ctx, "disable scheduled cleanup of s3", "fileExpireTime", c.config.CronTask.FileExpireTime, "deleteObjectType", c.config.CronTask.DeleteObjectType) + return nil + } + _, err := c.cron.AddFunc(c.config.CronTask.CronExecuteTime, c.clearS3) + return errs.WrapMsg(err, "failed to register clear s3 cron task") +} + +func (c *cronServer) registerDeleteMsg() error { + if c.config.CronTask.RetainChatRecords <= 0 { + log.ZInfo(c.ctx, "disable scheduled cleanup of chat records", "retainChatRecords", c.config.CronTask.RetainChatRecords) + return nil + } + _, err := c.cron.AddFunc(c.config.CronTask.CronExecuteTime, c.deleteMsg) + return errs.WrapMsg(err, "failed to register delete msg cron task") +} + +func (c *cronServer) registerClearUserMsg() error { + _, err := c.cron.AddFunc(c.config.CronTask.CronExecuteTime, c.clearUserMsg) + return errs.WrapMsg(err, "failed to register clear user msg cron task") +} diff --git a/internal/tools/msg.go b/internal/tools/msg.go new file mode 100644 index 000000000..1149b1f29 --- /dev/null +++ b/internal/tools/msg.go @@ -0,0 +1,23 @@ +package tools + +import ( + "fmt" + "github.com/openimsdk/protocol/msg" + "github.com/openimsdk/tools/log" + "github.com/openimsdk/tools/mcontext" + "os" + "time" +) + +func (c *cronServer) deleteMsg() { + now := time.Now() + deltime := now.Add(-time.Hour * 24 * time.Duration(c.config.CronTask.RetainChatRecords)) + ctx := mcontext.SetOperationID(c.ctx, fmt.Sprintf("cron_%d_%d", os.Getpid(), deltime.UnixMilli())) + log.ZDebug(ctx, "Destruct chat records", "deltime", deltime, "timestamp", deltime.UnixMilli()) + + if _, err := c.msgClient.DestructMsgs(ctx, &msg.DestructMsgsReq{Timestamp: deltime.UnixMilli()}); err != nil { + log.ZError(ctx, "cron destruct chat records failed", err, "deltime", deltime, "cont", time.Since(now)) + return + } + log.ZDebug(ctx, "cron destruct chat records success", "deltime", deltime, "cont", time.Since(now)) +} diff --git a/internal/tools/s3.go b/internal/tools/s3.go new file mode 100644 index 000000000..38d8badc6 --- /dev/null +++ b/internal/tools/s3.go @@ -0,0 +1,34 @@ +package tools + +import ( + "fmt" + "github.com/openimsdk/protocol/third" + "github.com/openimsdk/tools/log" + "github.com/openimsdk/tools/mcontext" + "os" + "time" +) + +func (c *cronServer) clearS3() { + start := time.Now() + executeNum := 10 + // number of pagination. if need modify, need update value in third.DeleteOutdatedData + pageShowNumber := 500 + deleteTime := start.Add(-time.Hour * 24 * time.Duration(c.config.CronTask.FileExpireTime)) + operationID := fmt.Sprintf("cron_%d_%d", os.Getpid(), deleteTime.UnixMilli()) + ctx := mcontext.SetOperationID(c.ctx, operationID) + log.ZDebug(ctx, "deleteoutDatedData", "deletetime", deleteTime, "timestamp", deleteTime.UnixMilli()) + for i := 1; i <= executeNum; i++ { + ctx := mcontext.SetOperationID(c.ctx, fmt.Sprintf("%s_%d", operationID, i)) + resp, err := c.thirdClient.DeleteOutdatedData(ctx, &third.DeleteOutdatedDataReq{ExpireTime: deleteTime.UnixMilli(), ObjectGroup: c.config.CronTask.DeleteObjectType, Count: int32(pageShowNumber)}) + if err != nil { + log.ZError(ctx, "cron deleteoutDatedData failed", err, "deleteTime", deleteTime, "cont", time.Since(start)) + return + } + if resp.Count < int32(pageShowNumber) { + break + } + } + + log.ZDebug(ctx, "cron deleteoutDatedData success", "deltime", deleteTime, "cont", time.Since(start)) +} diff --git a/internal/tools/user_msg.go b/internal/tools/user_msg.go new file mode 100644 index 000000000..9dfaf58c3 --- /dev/null +++ b/internal/tools/user_msg.go @@ -0,0 +1,31 @@ +package tools + +import ( + "fmt" + pbconversation "github.com/openimsdk/protocol/conversation" + "github.com/openimsdk/protocol/msg" + "github.com/openimsdk/tools/log" + "github.com/openimsdk/tools/mcontext" + "os" + "time" +) + +func (c *cronServer) clearUserMsg() { + now := time.Now() + ctx := mcontext.SetOperationID(c.ctx, fmt.Sprintf("cron_%d_%d", os.Getpid(), now.UnixMilli())) + log.ZDebug(ctx, "clear msg cron start", "now", now) + + conversations, err := c.conversationClient.GetConversationsNeedClearMsg(ctx, &pbconversation.GetConversationsNeedClearMsgReq{}) + if err != nil { + log.ZError(ctx, "Get conversation need Destruct msgs failed.", err) + return + } + + _, err = c.msgClient.ClearMsg(ctx, &msg.ClearMsgReq{Conversations: conversations.Conversations}) + if err != nil { + log.ZError(ctx, "Clear Msg failed.", err) + return + } + + log.ZDebug(ctx, "clear msg cron task completed", "cont", time.Since(now)) +} diff --git a/pkg/common/storage/controller/s3.go b/pkg/common/storage/controller/s3.go index 4e5ad18b6..6a0f11b1c 100644 --- a/pkg/common/storage/controller/s3.go +++ b/pkg/common/storage/controller/s3.go @@ -24,7 +24,6 @@ import ( "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model" "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache" - "github.com/openimsdk/tools/db/pagination" "github.com/openimsdk/tools/s3" "github.com/openimsdk/tools/s3/cont" "github.com/redis/go-redis/v9" @@ -40,10 +39,8 @@ type S3Database interface { SetObject(ctx context.Context, info *model.Object) error StatObject(ctx context.Context, name string) (*s3.ObjectInfo, error) FormData(ctx context.Context, name string, size int64, contentType string, duration time.Duration) (*s3.FormData, error) - FindNeedDeleteObjectByDB(ctx context.Context, duration time.Time, needDelType []string, pagination pagination.Pagination) (total int64, objects []*model.Object, err error) - DeleteObject(ctx context.Context, name string) error - DeleteSpecifiedData(ctx context.Context, engine string, name string) error - FindModelsByKey(ctx context.Context, key string) (objects []*model.Object, err error) + FindExpirationObject(ctx context.Context, engine string, expiration time.Time, needDelType []string, count int64) ([]*model.Object, error) + DeleteSpecifiedData(ctx context.Context, engine string, name []string) error DelS3Key(ctx context.Context, engine string, keys ...string) error } @@ -120,21 +117,14 @@ func (s *s3Database) StatObject(ctx context.Context, name string) (*s3.ObjectInf func (s *s3Database) FormData(ctx context.Context, name string, size int64, contentType string, duration time.Duration) (*s3.FormData, error) { return s.s3.FormData(ctx, name, size, contentType, duration) } -func (s *s3Database) FindNeedDeleteObjectByDB(ctx context.Context, duration time.Time, needDelType []string, pagination pagination.Pagination) (total int64, objects []*model.Object, err error) { - return s.db.FindNeedDeleteObjectByDB(ctx, duration, needDelType, pagination) +func (s *s3Database) FindExpirationObject(ctx context.Context, engine string, expiration time.Time, needDelType []string, count int64) ([]*model.Object, error) { + return s.db.FindExpirationObject(ctx, engine, expiration, needDelType, count) } -func (s *s3Database) DeleteObject(ctx context.Context, name string) error { - return s.s3.DeleteObject(ctx, name) -} -func (s *s3Database) DeleteSpecifiedData(ctx context.Context, engine string, name string) error { +func (s *s3Database) DeleteSpecifiedData(ctx context.Context, engine string, name []string) error { return s.db.Delete(ctx, engine, name) } -func (s *s3Database) FindModelsByKey(ctx context.Context, key string) (objects []*model.Object, err error) { - return s.db.FindModelsByKey(ctx, key) -} - func (s *s3Database) DelS3Key(ctx context.Context, engine string, keys ...string) error { return s.s3cache.DelS3Key(ctx, engine, keys...) } diff --git a/pkg/common/storage/database/mgo/object.go b/pkg/common/storage/database/mgo/object.go index 5bc329d33..53f2a6ba4 100644 --- a/pkg/common/storage/database/mgo/object.go +++ b/pkg/common/storage/database/mgo/object.go @@ -22,7 +22,6 @@ import ( "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model" "github.com/openimsdk/tools/db/mongoutil" - "github.com/openimsdk/tools/db/pagination" "github.com/openimsdk/tools/errs" "go.mongodb.org/mongo-driver/bson" "go.mongodb.org/mongo-driver/mongo" @@ -91,21 +90,21 @@ func (o *S3Mongo) Take(ctx context.Context, engine string, name string) (*model. return mongoutil.FindOne[*model.Object](ctx, o.coll, bson.M{"name": name, "engine": engine}) } -func (o *S3Mongo) Delete(ctx context.Context, engine string, name string) error { - return mongoutil.DeleteOne(ctx, o.coll, bson.M{"name": name, "engine": engine}) +func (o *S3Mongo) Delete(ctx context.Context, engine string, name []string) error { + if len(name) == 0 { + return nil + } + return mongoutil.DeleteOne(ctx, o.coll, bson.M{"engine": engine, "name": bson.M{"$in": name}}) } -// Find Expires object -func (o *S3Mongo) FindNeedDeleteObjectByDB(ctx context.Context, duration time.Time, needDelType []string, pagination pagination.Pagination) (total int64, objects []*model.Object, err error) { - return mongoutil.FindPage[*model.Object](ctx, o.coll, bson.M{ - "create_time": bson.M{"$lt": duration}, - "group": bson.M{"$in": needDelType}, - }, pagination) -} - -// Find object by key -func (o *S3Mongo) FindModelsByKey(ctx context.Context, key string) (objects []*model.Object, err error) { +func (o *S3Mongo) FindExpirationObject(ctx context.Context, engine string, expiration time.Time, needDelType []string, count int64) ([]*model.Object, error) { + opt := options.Find() + if count > 0 { + opt.SetLimit(count) + } return mongoutil.Find[*model.Object](ctx, o.coll, bson.M{ - "key": key, - }) + "engine": engine, + "create_time": bson.M{"$lt": expiration}, + "group": bson.M{"$in": needDelType}, + }, opt) } diff --git a/pkg/common/storage/database/object.go b/pkg/common/storage/database/object.go index c741e39a6..86d47a2a6 100644 --- a/pkg/common/storage/database/object.go +++ b/pkg/common/storage/database/object.go @@ -19,13 +19,11 @@ import ( "time" "github.com/openimsdk/open-im-server/v3/pkg/common/storage/model" - "github.com/openimsdk/tools/db/pagination" ) type ObjectInfo interface { SetObject(ctx context.Context, obj *model.Object) error Take(ctx context.Context, engine string, name string) (*model.Object, error) - Delete(ctx context.Context, engine string, name string) error - FindNeedDeleteObjectByDB(ctx context.Context, duration time.Time, needDelType []string, pagination pagination.Pagination) (total int64, objects []*model.Object, err error) - FindModelsByKey(ctx context.Context, key string) (objects []*model.Object, err error) + Delete(ctx context.Context, engine string, name []string) error + FindExpirationObject(ctx context.Context, engine string, expiration time.Time, needDelType []string, count int64) ([]*model.Object, error) }