mirror of
https://github.com/openimsdk/open-im-server.git
synced 2026-05-01 07:35:58 +08:00
Merge remote-tracking branch 'origin/errcode' into errcode
This commit is contained in:
withchao
@@ -0,0 +1,423 @@
|
||||
package controller
|
||||
|
||||
import "C"
|
||||
import (
|
||||
"OpenIM/pkg/common/db/obj"
|
||||
"OpenIM/pkg/common/db/table/relation"
|
||||
"OpenIM/pkg/common/tracelog"
|
||||
"OpenIM/pkg/proto/third"
|
||||
"OpenIM/pkg/utils"
|
||||
"context"
|
||||
"crypto/md5"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"fmt"
|
||||
"github.com/google/uuid"
|
||||
"path"
|
||||
"strconv"
|
||||
"time"
|
||||
)
|
||||
|
||||
type S3Database interface {
|
||||
ApplyPut(ctx context.Context, req *third.ApplyPutReq) (*third.ApplyPutResp, error)
|
||||
GetPut(ctx context.Context, req *third.GetPutReq) (*third.GetPutResp, error)
|
||||
ConfirmPut(ctx context.Context, req *third.ConfirmPutReq) (*third.ConfirmPutResp, error)
|
||||
}
|
||||
|
||||
func NewS3Database(obj obj.Interface, hash relation.ObjectHashModelInterface, info relation.ObjectInfoModelInterface, put relation.ObjectPutModelInterface) S3Database {
|
||||
return &s3Database{
|
||||
obj: obj,
|
||||
hash: hash,
|
||||
info: info,
|
||||
put: put,
|
||||
}
|
||||
}
|
||||
|
||||
type s3Database struct {
|
||||
obj obj.Interface
|
||||
hash relation.ObjectHashModelInterface
|
||||
info relation.ObjectInfoModelInterface
|
||||
put relation.ObjectPutModelInterface
|
||||
}
|
||||
|
||||
// today 今天的日期
|
||||
func (c *s3Database) today() string {
|
||||
return time.Now().Format("20060102")
|
||||
}
|
||||
|
||||
// fragmentName 根据序号生成文件名
|
||||
func (c *s3Database) fragmentName(index int) string {
|
||||
return "fragment_" + strconv.Itoa(index+1)
|
||||
}
|
||||
|
||||
// getFragmentNum 获取分片大小和分片数量
|
||||
func (c *s3Database) getFragmentNum(fragmentSize int64, objectSize int64) (int64, int) {
|
||||
if size := c.obj.MinFragmentSize(); fragmentSize < size {
|
||||
fragmentSize = size
|
||||
}
|
||||
if fragmentSize <= 0 || objectSize <= fragmentSize {
|
||||
return objectSize, 1
|
||||
} else {
|
||||
num := int(objectSize / fragmentSize)
|
||||
if objectSize%fragmentSize > 0 {
|
||||
num++
|
||||
}
|
||||
if n := c.obj.MaxFragmentNum(); num > n {
|
||||
num = n
|
||||
}
|
||||
return fragmentSize, num
|
||||
}
|
||||
}
|
||||
|
||||
func (c *s3Database) CheckHash(hash string) error {
|
||||
val, err := hex.DecodeString(hash)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if len(val) != md5.Size {
|
||||
return errors.New("hash value error")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *s3Database) urlName(name string) string {
|
||||
if name[0] != '/' {
|
||||
name = "/" + name
|
||||
}
|
||||
return "http://127.0.0.1:8080" + name
|
||||
}
|
||||
|
||||
func (c *s3Database) UUID() string {
|
||||
return uuid.New().String()
|
||||
}
|
||||
|
||||
func (c *s3Database) HashName(hash string) string {
|
||||
return path.Join("hash", c.today(), c.UUID())
|
||||
}
|
||||
|
||||
func (c *s3Database) isNotFound(err error) bool {
|
||||
return relation.IsNotFound(err)
|
||||
}
|
||||
|
||||
func (c *s3Database) ApplyPut(ctx context.Context, req *third.ApplyPutReq) (*third.ApplyPutResp, error) {
|
||||
if err := c.CheckHash(req.Hash); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := c.obj.CheckName(req.Name); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if req.CleanTime != 0 && req.CleanTime <= time.Now().UnixMilli() {
|
||||
return nil, errors.New("invalid CleanTime")
|
||||
}
|
||||
var expirationTime *time.Time
|
||||
if req.CleanTime != 0 {
|
||||
expirationTime = utils.ToPtr(time.UnixMilli(req.CleanTime))
|
||||
}
|
||||
if hash, err := c.hash.Take(ctx, req.Hash, c.obj.Name()); err == nil {
|
||||
o := relation.ObjectInfoModel{
|
||||
Name: req.Name,
|
||||
Hash: hash.Hash,
|
||||
ExpirationTime: expirationTime,
|
||||
CreateTime: time.Now(),
|
||||
}
|
||||
if err := c.info.SetObject(ctx, &o); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &third.ApplyPutResp{Url: c.urlName(o.Name)}, nil // 服务器已存在
|
||||
} else if !c.isNotFound(err) {
|
||||
return nil, err
|
||||
}
|
||||
// 新上传
|
||||
var pack int
|
||||
const effective = time.Hour * 24 * 2
|
||||
req.FragmentSize, pack = c.getFragmentNum(req.FragmentSize, req.Size)
|
||||
put := relation.ObjectPutModel{
|
||||
PutID: c.UUID(),
|
||||
Hash: req.Hash,
|
||||
Name: req.Name,
|
||||
ObjectSize: req.Size,
|
||||
FragmentSize: req.FragmentSize,
|
||||
ExpirationTime: expirationTime,
|
||||
EffectiveTime: time.Now().Add(effective),
|
||||
}
|
||||
put.Path = path.Join("upload", c.today(), req.Hash, put.PutID)
|
||||
putURLs := make([]string, 0, pack)
|
||||
for i := 0; i < pack; i++ {
|
||||
url, err := c.obj.PresignedPutURL(ctx, &obj.ApplyPutArgs{
|
||||
Bucket: c.obj.TempBucket(),
|
||||
Name: path.Join(put.Path, c.fragmentName(i)),
|
||||
Effective: effective,
|
||||
MaxObjectSize: req.FragmentSize,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
putURLs = append(putURLs, url)
|
||||
}
|
||||
put.CreateTime = time.Now()
|
||||
if err := c.put.Create(ctx, []*relation.ObjectPutModel{&put}); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &third.ApplyPutResp{
|
||||
PutID: put.PutID,
|
||||
FragmentSize: put.FragmentSize,
|
||||
PutURLs: putURLs,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (c *s3Database) GetPut(ctx context.Context, req *third.GetPutReq) (*third.GetPutResp, error) {
|
||||
up, err := c.put.Take(ctx, req.PutID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if up.Complete {
|
||||
return nil, errors.New("up completed")
|
||||
}
|
||||
_, pack := c.getFragmentNum(up.FragmentSize, up.ObjectSize)
|
||||
fragments := make([]*third.GetPutFragment, pack)
|
||||
for i := 0; i < pack; i++ {
|
||||
name := path.Join(up.Path, c.fragmentName(i))
|
||||
o, err := c.obj.GetObjectInfo(ctx, &obj.BucketObject{
|
||||
Bucket: c.obj.TempBucket(),
|
||||
Name: name,
|
||||
})
|
||||
if err != nil {
|
||||
if c.obj.IsNotFound(err) {
|
||||
fragments[i] = &third.GetPutFragment{}
|
||||
continue
|
||||
}
|
||||
return nil, err
|
||||
}
|
||||
fragments[i] = &third.GetPutFragment{Size: o.Size, Hash: o.Hash}
|
||||
}
|
||||
var cleanTime int64
|
||||
if up.ExpirationTime != nil {
|
||||
cleanTime = up.ExpirationTime.UnixMilli()
|
||||
}
|
||||
return &third.GetPutResp{
|
||||
FragmentSize: up.FragmentSize,
|
||||
Size: up.ObjectSize,
|
||||
Name: up.Name,
|
||||
Hash: up.Hash,
|
||||
Fragments: fragments,
|
||||
CleanTime: cleanTime,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (c *s3Database) ConfirmPut(ctx context.Context, req *third.ConfirmPutReq) (_ *third.ConfirmPutResp, _err error) {
|
||||
put, err := c.put.Take(ctx, req.PutID)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
_, pack := c.getFragmentNum(put.FragmentSize, put.ObjectSize)
|
||||
defer func() {
|
||||
if _err == nil {
|
||||
// 清理上传的碎片
|
||||
err := c.obj.DeleteObjet(ctx, &obj.BucketObject{Bucket: c.obj.TempBucket(), Name: put.Path})
|
||||
if err != nil {
|
||||
tracelog.SetCtxWarn(ctx, "DeleteObjet", err, "Bucket", c.obj.TempBucket(), "Path", put.Path)
|
||||
}
|
||||
}
|
||||
}()
|
||||
if put.Complete {
|
||||
return nil, errors.New("put completed")
|
||||
}
|
||||
now := time.Now().UnixMilli()
|
||||
if put.EffectiveTime.UnixMilli() < now {
|
||||
return nil, errors.New("upload expired")
|
||||
}
|
||||
if put.ExpirationTime != nil && put.ExpirationTime.UnixMilli() < now {
|
||||
return nil, errors.New("object expired")
|
||||
}
|
||||
if hash, err := c.hash.Take(ctx, put.Hash, c.obj.Name()); err == nil {
|
||||
o := relation.ObjectInfoModel{
|
||||
Name: put.Name,
|
||||
Hash: hash.Hash,
|
||||
ExpirationTime: put.ExpirationTime,
|
||||
CreateTime: time.Now(),
|
||||
}
|
||||
if err := c.info.SetObject(ctx, &o); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer func() {
|
||||
err := c.obj.DeleteObjet(ctx, &obj.BucketObject{
|
||||
Bucket: c.obj.TempBucket(),
|
||||
Name: put.Path,
|
||||
})
|
||||
if err != nil {
|
||||
tracelog.SetCtxWarn(ctx, "DeleteObjet", err, "Bucket", c.obj.TempBucket(), "Path", put.Path)
|
||||
}
|
||||
}()
|
||||
// 服务端已存在
|
||||
return &third.ConfirmPutResp{
|
||||
Url: c.urlName(o.Name),
|
||||
}, nil
|
||||
} else if c.isNotFound(err) {
|
||||
return nil, err
|
||||
}
|
||||
src := make([]obj.BucketObject, pack)
|
||||
for i := 0; i < pack; i++ {
|
||||
name := path.Join(put.Path, c.fragmentName(i))
|
||||
o, err := c.obj.GetObjectInfo(ctx, &obj.BucketObject{
|
||||
Bucket: c.obj.TempBucket(),
|
||||
Name: name,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if i+1 == pack { // 最后一个
|
||||
size := put.ObjectSize - put.FragmentSize*int64(i)
|
||||
if size != o.Size {
|
||||
return nil, fmt.Errorf("last fragment %d size %d not equal to %d hash %s", i, o.Size, size, o.Hash)
|
||||
}
|
||||
} else {
|
||||
if o.Size != put.FragmentSize {
|
||||
return nil, fmt.Errorf("fragment %d size %d not equal to %d hash %s", i, o.Size, put.FragmentSize, o.Hash)
|
||||
}
|
||||
}
|
||||
src[i] = obj.BucketObject{
|
||||
Bucket: c.obj.TempBucket(),
|
||||
Name: name,
|
||||
}
|
||||
}
|
||||
dst := &obj.BucketObject{
|
||||
Bucket: c.obj.DataBucket(),
|
||||
Name: c.HashName(put.Hash),
|
||||
}
|
||||
if len(src) == 1 { // 未分片直接触发copy
|
||||
// 检查数据完整性,避免脏数据
|
||||
o, err := c.obj.GetObjectInfo(ctx, &src[0])
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if put.ObjectSize != o.Size {
|
||||
return nil, fmt.Errorf("size mismatching should %d reality %d", put.ObjectSize, o.Size)
|
||||
}
|
||||
if put.Hash != o.Hash {
|
||||
return nil, fmt.Errorf("hash mismatching should %s reality %s", put.Hash, o.Hash)
|
||||
}
|
||||
if err := c.obj.CopyObjet(ctx, &src[0], dst); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
} else {
|
||||
tempBucket := &obj.BucketObject{
|
||||
Bucket: c.obj.TempBucket(),
|
||||
Name: path.Join(put.Path, "merge_"+c.UUID()),
|
||||
}
|
||||
defer func() { // 清理合成的文件
|
||||
if err := c.obj.DeleteObjet(ctx, tempBucket); err != nil {
|
||||
tracelog.SetCtxWarn(ctx, "DeleteObjet", err, "Bucket", tempBucket.Bucket, "Path", tempBucket.Name)
|
||||
}
|
||||
}()
|
||||
err := c.obj.ComposeObject(ctx, src, tempBucket)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
info, err := c.obj.GetObjectInfo(ctx, tempBucket)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if put.ObjectSize != info.Size {
|
||||
return nil, fmt.Errorf("size mismatch should %d reality %d", put.ObjectSize, info.Size)
|
||||
}
|
||||
if put.Hash != info.Hash {
|
||||
return nil, fmt.Errorf("hash mismatch should %s reality %s", put.Hash, info.Hash)
|
||||
}
|
||||
if err := c.obj.CopyObjet(ctx, tempBucket, dst); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
h := &relation.ObjectHashModel{
|
||||
Hash: put.Hash,
|
||||
Engine: c.obj.Name(),
|
||||
Size: put.ObjectSize,
|
||||
Bucket: c.obj.DataBucket(),
|
||||
Name: dst.Name,
|
||||
CreateTime: time.Now(),
|
||||
}
|
||||
o := &relation.ObjectInfoModel{
|
||||
Name: put.Name,
|
||||
Hash: put.Hash,
|
||||
ExpirationTime: put.ExpirationTime,
|
||||
CreateTime: time.Now(),
|
||||
}
|
||||
if err := c.hash.Create(ctx, []*relation.ObjectHashModel{h}); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := c.info.SetObject(ctx, o); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if err := c.put.SetCompleted(ctx, put.PutID); err != nil {
|
||||
tracelog.SetCtxWarn(ctx, "SetCompleted", err, "PutID", put.PutID)
|
||||
}
|
||||
return &third.ConfirmPutResp{
|
||||
Url: c.urlName(o.Name),
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (c *s3Database) CleanExpirationObject(ctx context.Context, t time.Time) {
|
||||
// 清理上传产生的临时文件
|
||||
c.cleanPutTemp(ctx, t, 10)
|
||||
// 清理hash引用全过期的文件
|
||||
c.cleanExpirationObject(ctx, t)
|
||||
// 清理没有引用的hash对象
|
||||
c.clearNoCitation(ctx, c.obj.Name(), 10)
|
||||
}
|
||||
|
||||
func (c *s3Database) cleanPutTemp(ctx context.Context, t time.Time, num int) {
|
||||
for {
|
||||
puts, err := c.put.FindExpirationPut(ctx, t, num)
|
||||
if err != nil {
|
||||
tracelog.SetCtxWarn(ctx, "FindExpirationPut", err, "Time", t, "Num", num)
|
||||
return
|
||||
}
|
||||
if len(puts) == 0 {
|
||||
return
|
||||
}
|
||||
for _, put := range puts {
|
||||
err := c.obj.DeleteObjet(ctx, &obj.BucketObject{Bucket: c.obj.TempBucket(), Name: put.Path})
|
||||
if err != nil {
|
||||
tracelog.SetCtxWarn(ctx, "DeleteObjet", err, "Bucket", c.obj.TempBucket(), "Path", put.Path)
|
||||
return
|
||||
}
|
||||
}
|
||||
ids := utils.Slice(puts, func(e *relation.ObjectPutModel) string { return e.PutID })
|
||||
err = c.put.DelPut(ctx, ids)
|
||||
if err != nil {
|
||||
tracelog.SetCtxWarn(ctx, "DelPut", err, "PutID", ids)
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *s3Database) cleanExpirationObject(ctx context.Context, t time.Time) {
|
||||
err := c.info.DeleteExpiration(ctx, t)
|
||||
if err != nil {
|
||||
tracelog.SetCtxWarn(ctx, "DeleteExpiration", err, "Time", t)
|
||||
}
|
||||
}
|
||||
|
||||
func (c *s3Database) clearNoCitation(ctx context.Context, engine string, limit int) {
|
||||
for {
|
||||
list, err := c.hash.DeleteNoCitation(ctx, engine, limit)
|
||||
if err != nil {
|
||||
tracelog.SetCtxWarn(ctx, "DeleteNoCitation", err, "Engine", engine, "Limit", limit)
|
||||
return
|
||||
}
|
||||
if len(list) == 0 {
|
||||
return
|
||||
}
|
||||
var hasErr bool
|
||||
for _, h := range list {
|
||||
err := c.obj.DeleteObjet(ctx, &obj.BucketObject{Bucket: h.Bucket, Name: h.Name})
|
||||
if err != nil {
|
||||
hasErr = true
|
||||
tracelog.SetCtxWarn(ctx, "DeleteObjet", err, "Bucket", h.Bucket, "Path", h.Name)
|
||||
continue
|
||||
}
|
||||
}
|
||||
if hasErr {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user