708 lines
23 KiB
Go
708 lines
23 KiB
Go
// 对象存储:namespace / 桶 / 对象 / 预签名请求(PAR)。
|
|
// 上传下载数据面走 PAR 直连 OCI,面板只做控制面;
|
|
// 例外:预览/编辑的小文件经面板中转(GetObject/PutObject),不签发 PAR。
|
|
package oci
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/oracle/oci-go-sdk/v65/common"
|
|
"github.com/oracle/oci-go-sdk/v65/objectstorage"
|
|
)
|
|
|
|
// ErrObjectTooLarge 表示对象超出面板中转大小上限。
|
|
var ErrObjectTooLarge = errors.New("object too large for inline transfer")
|
|
|
|
// ObjectContent 是对象内容与写入所需元数据(面板中转小文件用)。
|
|
type ObjectContent struct {
|
|
Data []byte
|
|
ContentType string
|
|
Etag string
|
|
}
|
|
|
|
// Bucket 是存储桶摘要(含用量估算,来自 GetBucket 的 approximate 字段)。
|
|
type Bucket struct {
|
|
Name string `json:"name"`
|
|
Namespace string `json:"namespace"`
|
|
Visibility string `json:"visibility"` // NoPublicAccess / ObjectRead
|
|
StorageTier string `json:"storageTier"`
|
|
VersioningOn bool `json:"versioningOn"`
|
|
ApproximateCount int64 `json:"approximateCount"`
|
|
ApproximateSize int64 `json:"approximateSize"`
|
|
TimeCreated *time.Time `json:"timeCreated"`
|
|
}
|
|
|
|
// ObjectSummary 是对象列表行;文件夹以 prefixes 单独返回。
|
|
type ObjectSummary struct {
|
|
Name string `json:"name"`
|
|
Size int64 `json:"size"`
|
|
StorageTier string `json:"storageTier"`
|
|
ArchivalState string `json:"archivalState"`
|
|
TimeModified *time.Time `json:"timeModified"`
|
|
}
|
|
|
|
// ListObjectsResult 是分页对象列表:delimiter="/" 模式,子层级归入 Prefixes。
|
|
type ListObjectsResult struct {
|
|
Objects []ObjectSummary `json:"objects"`
|
|
Prefixes []string `json:"prefixes"`
|
|
NextStartWith string `json:"nextStartWith"`
|
|
}
|
|
|
|
// ObjectDetail 是对象元数据(HEAD)。
|
|
type ObjectDetail struct {
|
|
Name string `json:"name"`
|
|
Size int64 `json:"size"`
|
|
ContentType string `json:"contentType"`
|
|
Etag string `json:"etag"`
|
|
ContentMd5 string `json:"contentMd5"`
|
|
StorageTier string `json:"storageTier"`
|
|
ArchivalState string `json:"archivalState"`
|
|
TimeModified *time.Time `json:"timeModified"`
|
|
Metadata map[string]string `json:"metadata"`
|
|
}
|
|
|
|
// PAR 是预签名请求摘要;FullURL 仅创建响应携带(OCI 事后不可再取回)。
|
|
type PAR struct {
|
|
ID string `json:"id"`
|
|
Name string `json:"name"`
|
|
ObjectName string `json:"objectName"`
|
|
AccessType string `json:"accessType"`
|
|
FullURL string `json:"fullUrl,omitempty"`
|
|
TimeExpires *time.Time `json:"timeExpires"`
|
|
TimeCreated *time.Time `json:"timeCreated"`
|
|
}
|
|
|
|
// CreateBucketInput 创建桶参数;CompartmentID 空为生效 compartment。
|
|
type CreateBucketInput struct {
|
|
Name string
|
|
CompartmentID string
|
|
PublicRead bool
|
|
StorageTier string // Standard / Archive
|
|
VersioningOn bool
|
|
}
|
|
|
|
// UpdateBucketInput 更新桶参数;nil 字段不改。
|
|
type UpdateBucketInput struct {
|
|
PublicRead *bool
|
|
VersioningOn *bool
|
|
}
|
|
|
|
// CreatePARInput 签发预签名请求参数。
|
|
type CreatePARInput struct {
|
|
Name string
|
|
ObjectName string // 空 = 桶级(配合 AnyObjectReadWrite 前缀用法)
|
|
AccessType string // ObjectRead / ObjectWrite / ObjectReadWrite / AnyObjectReadWrite
|
|
ExpiresHours int
|
|
}
|
|
|
|
// bulkRetryPolicy 批量删除(清空桶)的限流退避:16 并发下 OCI 可能回 429。
|
|
// 不用 DefaultRetryPolicy:其 8 次尝试、退避上限 30s,且带最终一致性模式
|
|
// (9 次、上限 45s)——大批量删除里 worker 一旦触发就长睡,吞吐塌方;
|
|
// 这里收紧为 5 次、上限 3s、关掉最终一致性,单条最坏 ~13s 后放行不拖全场。
|
|
var bulkRetryPolicy = common.NewRetryPolicyWithOptions(
|
|
common.ReplaceWithValuesFromRetryPolicy(common.DefaultRetryPolicyWithoutEventualConsistency()),
|
|
common.WithMaximumNumberAttempts(5),
|
|
common.WithExponentialBackoff(3*time.Second, 2.0),
|
|
)
|
|
|
|
func (c *RealClient) osClient(cred Credentials, region string) (objectstorage.ObjectStorageClient, error) {
|
|
oc, err := objectstorage.NewObjectStorageClientWithConfigurationProvider(provider(cred))
|
|
if err != nil {
|
|
return oc, fmt.Errorf("new object storage client: %w", err)
|
|
}
|
|
applyProxy(&oc.BaseClient, cred)
|
|
if region != "" {
|
|
oc.SetRegion(normalizeRegion(region))
|
|
}
|
|
return oc, nil
|
|
}
|
|
|
|
// GetObjectStorageNamespace 实现 Client:租户 namespace(全租户唯一,常量语义),
|
|
// 首次回源后按 tenancy 进程内缓存;此前每个对象存储操作都远程取一次,
|
|
// 经代理链路时白付一整次冷连接往返。并发首次可能重复回源,结果相同无害。
|
|
func (c *RealClient) GetObjectStorageNamespace(ctx context.Context, cred Credentials, region string) (string, error) {
|
|
if v, ok := c.namespaces.Load(cred.TenancyOCID); ok {
|
|
return v.(string), nil
|
|
}
|
|
oc, err := c.osClient(cred, region)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
resp, err := oc.GetNamespace(ctx, objectstorage.GetNamespaceRequest{})
|
|
if err != nil {
|
|
return "", fmt.Errorf("get namespace: %w", err)
|
|
}
|
|
if cred.TenancyOCID != "" {
|
|
c.namespaces.Store(cred.TenancyOCID, deref(resp.Value))
|
|
}
|
|
return deref(resp.Value), nil
|
|
}
|
|
|
|
// ListBuckets 实现 Client:列出指定 compartment 下的桶并逐个补全用量(空为生效 compartment)。
|
|
func (c *RealClient) ListBuckets(ctx context.Context, cred Credentials, region, compartmentID string) ([]Bucket, error) {
|
|
oc, err := c.osClient(cred, region)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
ns, err := c.GetObjectStorageNamespace(ctx, cred, region)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
resp, err := oc.ListBuckets(ctx, objectstorage.ListBucketsRequest{
|
|
NamespaceName: &ns,
|
|
CompartmentId: hostCompartment(cred, compartmentID),
|
|
})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("list buckets: %w", err)
|
|
}
|
|
out := make([]Bucket, 0, len(resp.Items))
|
|
for _, it := range resp.Items {
|
|
out = append(out, c.bucketDetail(ctx, oc, ns, deref(it.Name)))
|
|
}
|
|
return out, nil
|
|
}
|
|
|
|
// bucketDetail 取单桶详情;失败时退化为只有名字的摘要,不拖垮列表。
|
|
func (c *RealClient) bucketDetail(ctx context.Context, oc objectstorage.ObjectStorageClient, ns, name string) Bucket {
|
|
resp, err := oc.GetBucket(ctx, objectstorage.GetBucketRequest{
|
|
NamespaceName: &ns,
|
|
BucketName: &name,
|
|
Fields: []objectstorage.GetBucketFieldsEnum{"approximateCount", "approximateSize"},
|
|
})
|
|
if err != nil {
|
|
return Bucket{Name: name, Namespace: ns}
|
|
}
|
|
b := Bucket{
|
|
Name: name,
|
|
Namespace: ns,
|
|
Visibility: string(resp.PublicAccessType),
|
|
StorageTier: string(resp.StorageTier),
|
|
VersioningOn: resp.Versioning == objectstorage.BucketVersioningEnabled,
|
|
ApproximateCount: derefI64(resp.ApproximateCount),
|
|
ApproximateSize: derefI64(resp.ApproximateSize),
|
|
}
|
|
if resp.TimeCreated != nil {
|
|
b.TimeCreated = &resp.TimeCreated.Time
|
|
}
|
|
return b
|
|
}
|
|
|
|
func derefI64(p *int64) int64 {
|
|
if p != nil {
|
|
return *p
|
|
}
|
|
return 0
|
|
}
|
|
|
|
// CreateBucket 实现 Client。
|
|
func (c *RealClient) CreateBucket(ctx context.Context, cred Credentials, region string, in CreateBucketInput) (Bucket, error) {
|
|
oc, err := c.osClient(cred, region)
|
|
if err != nil {
|
|
return Bucket{}, err
|
|
}
|
|
ns, err := c.GetObjectStorageNamespace(ctx, cred, region)
|
|
if err != nil {
|
|
return Bucket{}, err
|
|
}
|
|
details := objectstorage.CreateBucketDetails{
|
|
Name: &in.Name,
|
|
CompartmentId: hostCompartment(cred, in.CompartmentID),
|
|
PublicAccessType: objectstorage.CreateBucketDetailsPublicAccessTypeNopublicaccess,
|
|
StorageTier: objectstorage.CreateBucketDetailsStorageTierStandard,
|
|
}
|
|
if in.PublicRead {
|
|
details.PublicAccessType = objectstorage.CreateBucketDetailsPublicAccessTypeObjectread
|
|
}
|
|
if strings.EqualFold(in.StorageTier, "Archive") {
|
|
details.StorageTier = objectstorage.CreateBucketDetailsStorageTierArchive
|
|
}
|
|
if in.VersioningOn {
|
|
details.Versioning = objectstorage.CreateBucketDetailsVersioningEnabled
|
|
}
|
|
if _, err := oc.CreateBucket(ctx, objectstorage.CreateBucketRequest{
|
|
NamespaceName: &ns, CreateBucketDetails: details,
|
|
}); err != nil {
|
|
return Bucket{}, fmt.Errorf("create bucket: %w", err)
|
|
}
|
|
return c.bucketDetail(ctx, oc, ns, in.Name), nil
|
|
}
|
|
|
|
// UpdateBucket 实现 Client:改可见性 / 版本控制。
|
|
func (c *RealClient) UpdateBucket(ctx context.Context, cred Credentials, region, name string, in UpdateBucketInput) (Bucket, error) {
|
|
oc, err := c.osClient(cred, region)
|
|
if err != nil {
|
|
return Bucket{}, err
|
|
}
|
|
ns, err := c.GetObjectStorageNamespace(ctx, cred, region)
|
|
if err != nil {
|
|
return Bucket{}, err
|
|
}
|
|
details := objectstorage.UpdateBucketDetails{}
|
|
if in.PublicRead != nil {
|
|
details.PublicAccessType = objectstorage.UpdateBucketDetailsPublicAccessTypeNopublicaccess
|
|
if *in.PublicRead {
|
|
details.PublicAccessType = objectstorage.UpdateBucketDetailsPublicAccessTypeObjectread
|
|
}
|
|
}
|
|
if in.VersioningOn != nil {
|
|
details.Versioning = objectstorage.UpdateBucketDetailsVersioningSuspended
|
|
if *in.VersioningOn {
|
|
details.Versioning = objectstorage.UpdateBucketDetailsVersioningEnabled
|
|
}
|
|
}
|
|
if _, err := oc.UpdateBucket(ctx, objectstorage.UpdateBucketRequest{
|
|
NamespaceName: &ns, BucketName: &name, UpdateBucketDetails: details,
|
|
}); err != nil {
|
|
return Bucket{}, fmt.Errorf("update bucket: %w", err)
|
|
}
|
|
return c.bucketDetail(ctx, oc, ns, name), nil
|
|
}
|
|
|
|
// DeleteBucket 实现 Client:仅空桶可删,OCI 拒绝非空桶。
|
|
// ErrBucketNotEmpty 表示桶内仍有对象 / 历史版本 / 未完成分片上传,OCI 拒绝删除。
|
|
var ErrBucketNotEmpty = errors.New("bucket not empty")
|
|
|
|
func (c *RealClient) DeleteBucket(ctx context.Context, cred Credentials, region, name string) error {
|
|
oc, err := c.osClient(cred, region)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
ns, err := c.GetObjectStorageNamespace(ctx, cred, region)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if _, err := oc.DeleteBucket(ctx, objectstorage.DeleteBucketRequest{
|
|
NamespaceName: &ns, BucketName: &name,
|
|
}); err != nil {
|
|
if isBucketNotEmptyErr(err) {
|
|
return fmt.Errorf("delete bucket %s: %w", name, ErrBucketNotEmpty)
|
|
}
|
|
return fmt.Errorf("delete bucket: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// isBucketNotEmptyErr 识别「桶非空」类删除拒绝:对象/版本/分片报 BucketNotEmpty,
|
|
// 仅剩活跃 PAR 时 OCI 报 409 且消息为 Active Preauthenticated Requests still exist,
|
|
// 两者都可经后台清空(对象版本+PAR)后重删。
|
|
func isBucketNotEmptyErr(err error) bool {
|
|
var se common.ServiceError
|
|
if !errors.As(err, &se) {
|
|
return false
|
|
}
|
|
if se.GetCode() == "BucketNotEmpty" {
|
|
return true
|
|
}
|
|
return se.GetHTTPStatusCode() == 409 &&
|
|
strings.Contains(se.GetMessage(), "Preauthenticated Request")
|
|
}
|
|
|
|
// ObjectVersion 是对象版本摘要,清空桶时逐版本删除用。
|
|
type ObjectVersion struct {
|
|
Name string `json:"name"`
|
|
VersionID string `json:"versionId"`
|
|
}
|
|
|
|
// ListObjectVersions 实现 Client:分页列出全部对象版本(未开版本控制的桶返回当前版本)。
|
|
func (c *RealClient) ListObjectVersions(ctx context.Context, cred Credentials, region, bucket, page string) ([]ObjectVersion, string, error) {
|
|
oc, err := c.osClient(cred, region)
|
|
if err != nil {
|
|
return nil, "", err
|
|
}
|
|
ns, err := c.GetObjectStorageNamespace(ctx, cred, region)
|
|
if err != nil {
|
|
return nil, "", err
|
|
}
|
|
req := objectstorage.ListObjectVersionsRequest{
|
|
NamespaceName: &ns, BucketName: &bucket, Limit: common.Int(1000),
|
|
}
|
|
if page != "" {
|
|
req.Page = &page
|
|
}
|
|
resp, err := oc.ListObjectVersions(ctx, req)
|
|
if err != nil {
|
|
return nil, "", fmt.Errorf("list object versions: %w", err)
|
|
}
|
|
out := make([]ObjectVersion, 0, len(resp.Items))
|
|
for _, it := range resp.Items {
|
|
out = append(out, ObjectVersion{Name: deref(it.Name), VersionID: deref(it.VersionId)})
|
|
}
|
|
return out, deref(resp.OpcNextPage), nil
|
|
}
|
|
|
|
// DeleteObjectVersion 实现 Client:删除指定版本(versionID 空为当前版本)。
|
|
func (c *RealClient) DeleteObjectVersion(ctx context.Context, cred Credentials, region, bucket, object, versionID string) error {
|
|
oc, err := c.osClient(cred, region)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
ns, err := c.GetObjectStorageNamespace(ctx, cred, region)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
req := objectstorage.DeleteObjectRequest{
|
|
NamespaceName: &ns, BucketName: &bucket, ObjectName: &object,
|
|
RequestMetadata: common.RequestMetadata{RetryPolicy: &bulkRetryPolicy},
|
|
}
|
|
if versionID != "" {
|
|
req.VersionId = &versionID
|
|
}
|
|
if _, err := oc.DeleteObject(ctx, req); err != nil {
|
|
return fmt.Errorf("delete object version: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// ListObjects 实现 Client:delimiter="/" 前缀模式;startWith 为上页游标。
|
|
func (c *RealClient) ListObjects(ctx context.Context, cred Credentials, region, bucket, prefix, startWith string, limit int) (ListObjectsResult, error) {
|
|
oc, err := c.osClient(cred, region)
|
|
if err != nil {
|
|
return ListObjectsResult{}, err
|
|
}
|
|
ns, err := c.GetObjectStorageNamespace(ctx, cred, region)
|
|
if err != nil {
|
|
return ListObjectsResult{}, err
|
|
}
|
|
if limit <= 0 || limit > 500 {
|
|
limit = 100
|
|
}
|
|
req := objectstorage.ListObjectsRequest{
|
|
NamespaceName: &ns,
|
|
BucketName: &bucket,
|
|
Delimiter: common.String("/"),
|
|
Limit: &limit,
|
|
Fields: common.String("name,size,timeModified,storageTier,archivalState"),
|
|
}
|
|
if prefix != "" {
|
|
req.Prefix = &prefix
|
|
}
|
|
if startWith != "" {
|
|
req.Start = &startWith
|
|
}
|
|
resp, err := oc.ListObjects(ctx, req)
|
|
if err != nil {
|
|
return ListObjectsResult{}, fmt.Errorf("list objects: %w", err)
|
|
}
|
|
return toListObjectsResult(resp), nil
|
|
}
|
|
|
|
func toListObjectsResult(resp objectstorage.ListObjectsResponse) ListObjectsResult {
|
|
out := ListObjectsResult{
|
|
Objects: make([]ObjectSummary, 0, len(resp.Objects)),
|
|
Prefixes: orEmpty(resp.Prefixes),
|
|
NextStartWith: deref(resp.NextStartWith),
|
|
}
|
|
for _, o := range resp.Objects {
|
|
item := ObjectSummary{
|
|
Name: deref(o.Name),
|
|
Size: derefI64(o.Size),
|
|
StorageTier: string(o.StorageTier),
|
|
ArchivalState: string(o.ArchivalState),
|
|
}
|
|
if o.TimeModified != nil {
|
|
item.TimeModified = &o.TimeModified.Time
|
|
}
|
|
out.Objects = append(out.Objects, item)
|
|
}
|
|
return out
|
|
}
|
|
|
|
// DeleteObject 实现 Client。
|
|
func (c *RealClient) DeleteObject(ctx context.Context, cred Credentials, region, bucket, object string) error {
|
|
oc, err := c.osClient(cred, region)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
ns, err := c.GetObjectStorageNamespace(ctx, cred, region)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if _, err := oc.DeleteObject(ctx, objectstorage.DeleteObjectRequest{
|
|
NamespaceName: &ns, BucketName: &bucket, ObjectName: &object,
|
|
}); err != nil {
|
|
return fmt.Errorf("delete object: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// RenameObject 实现 Client:OCI 原生重命名(同桶内)。
|
|
func (c *RealClient) RenameObject(ctx context.Context, cred Credentials, region, bucket, src, dst string) error {
|
|
oc, err := c.osClient(cred, region)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
ns, err := c.GetObjectStorageNamespace(ctx, cred, region)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if _, err := oc.RenameObject(ctx, objectstorage.RenameObjectRequest{
|
|
NamespaceName: &ns,
|
|
BucketName: &bucket,
|
|
RenameObjectDetails: objectstorage.RenameObjectDetails{
|
|
SourceName: &src,
|
|
NewName: &dst,
|
|
},
|
|
}); err != nil {
|
|
return fmt.Errorf("rename object: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// RestoreObject 实现 Client:Archive 对象取回,hours 为可下载时长(默认 24)。
|
|
func (c *RealClient) RestoreObject(ctx context.Context, cred Credentials, region, bucket, object string, hours int) error {
|
|
oc, err := c.osClient(cred, region)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
ns, err := c.GetObjectStorageNamespace(ctx, cred, region)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
req := objectstorage.RestoreObjectsRequest{
|
|
NamespaceName: &ns,
|
|
BucketName: &bucket,
|
|
RestoreObjectsDetails: objectstorage.RestoreObjectsDetails{
|
|
ObjectName: &object,
|
|
},
|
|
}
|
|
if hours > 0 {
|
|
req.RestoreObjectsDetails.Hours = &hours
|
|
}
|
|
if _, err := oc.RestoreObjects(ctx, req); err != nil {
|
|
return fmt.Errorf("restore object: %w", err)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// HeadObject 实现 Client:对象元数据。
|
|
func (c *RealClient) HeadObject(ctx context.Context, cred Credentials, region, bucket, object string) (ObjectDetail, error) {
|
|
oc, err := c.osClient(cred, region)
|
|
if err != nil {
|
|
return ObjectDetail{}, err
|
|
}
|
|
ns, err := c.GetObjectStorageNamespace(ctx, cred, region)
|
|
if err != nil {
|
|
return ObjectDetail{}, err
|
|
}
|
|
resp, err := oc.HeadObject(ctx, objectstorage.HeadObjectRequest{
|
|
NamespaceName: &ns, BucketName: &bucket, ObjectName: &object,
|
|
})
|
|
if err != nil {
|
|
return ObjectDetail{}, fmt.Errorf("head object: %w", err)
|
|
}
|
|
d := ObjectDetail{
|
|
Name: object,
|
|
Size: derefI64(resp.ContentLength),
|
|
ContentType: deref(resp.ContentType),
|
|
Etag: deref(resp.ETag),
|
|
ContentMd5: deref(resp.ContentMd5),
|
|
StorageTier: string(resp.StorageTier),
|
|
ArchivalState: string(resp.ArchivalState),
|
|
Metadata: resp.OpcMeta,
|
|
}
|
|
if resp.LastModified != nil {
|
|
d.TimeModified = &resp.LastModified.Time
|
|
}
|
|
return d, nil
|
|
}
|
|
|
|
// GetObject 实现 Client:整体读入对象内容;超过 maxBytes 返回 ErrObjectTooLarge。
|
|
func (c *RealClient) GetObject(ctx context.Context, cred Credentials, region, bucket, object string, maxBytes int64) (ObjectContent, error) {
|
|
oc, err := c.osClient(cred, region)
|
|
if err != nil {
|
|
return ObjectContent{}, err
|
|
}
|
|
ns, err := c.GetObjectStorageNamespace(ctx, cred, region)
|
|
if err != nil {
|
|
return ObjectContent{}, err
|
|
}
|
|
resp, err := oc.GetObject(ctx, objectstorage.GetObjectRequest{
|
|
NamespaceName: &ns, BucketName: &bucket, ObjectName: &object,
|
|
})
|
|
if err != nil {
|
|
return ObjectContent{}, fmt.Errorf("get object: %w", err)
|
|
}
|
|
defer resp.Content.Close()
|
|
if derefI64(resp.ContentLength) > maxBytes {
|
|
return ObjectContent{}, fmt.Errorf("get object %s (%d bytes): %w", object, derefI64(resp.ContentLength), ErrObjectTooLarge)
|
|
}
|
|
data, err := io.ReadAll(io.LimitReader(resp.Content, maxBytes+1))
|
|
if err != nil {
|
|
return ObjectContent{}, fmt.Errorf("read object body: %w", err)
|
|
}
|
|
if int64(len(data)) > maxBytes {
|
|
return ObjectContent{}, fmt.Errorf("get object %s: %w", object, ErrObjectTooLarge)
|
|
}
|
|
return ObjectContent{Data: data, ContentType: deref(resp.ContentType), Etag: deref(resp.ETag)}, nil
|
|
}
|
|
|
|
// PutObject 实现 Client:整体写入对象;ifMatch 非空时作为 If-Match 条件,
|
|
// 不匹配由 OCI 返回 412(经 ServiceError 透出)。返回新 ETag。
|
|
func (c *RealClient) PutObject(ctx context.Context, cred Credentials, region, bucket, object string, data []byte, contentType, ifMatch string) (string, error) {
|
|
oc, err := c.osClient(cred, region)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
ns, err := c.GetObjectStorageNamespace(ctx, cred, region)
|
|
if err != nil {
|
|
return "", err
|
|
}
|
|
length := int64(len(data))
|
|
req := objectstorage.PutObjectRequest{
|
|
NamespaceName: &ns, BucketName: &bucket, ObjectName: &object,
|
|
ContentLength: &length,
|
|
PutObjectBody: io.NopCloser(bytes.NewReader(data)),
|
|
}
|
|
if contentType != "" {
|
|
req.ContentType = &contentType
|
|
}
|
|
if ifMatch != "" {
|
|
req.IfMatch = &ifMatch
|
|
}
|
|
resp, err := oc.PutObject(ctx, req)
|
|
if err != nil {
|
|
return "", fmt.Errorf("put object: %w", err)
|
|
}
|
|
return deref(resp.ETag), nil
|
|
}
|
|
|
|
// CreatePAR 实现 Client:签发预签名请求并拼出完整 URL(仅此时可得)。
|
|
func (c *RealClient) CreatePAR(ctx context.Context, cred Credentials, region, bucket string, in CreatePARInput) (PAR, error) {
|
|
oc, err := c.osClient(cred, region)
|
|
if err != nil {
|
|
return PAR{}, err
|
|
}
|
|
ns, err := c.GetObjectStorageNamespace(ctx, cred, region)
|
|
if err != nil {
|
|
return PAR{}, err
|
|
}
|
|
if in.ExpiresHours <= 0 {
|
|
in.ExpiresHours = 24
|
|
}
|
|
details := objectstorage.CreatePreauthenticatedRequestDetails{
|
|
Name: &in.Name,
|
|
AccessType: objectstorage.CreatePreauthenticatedRequestDetailsAccessTypeEnum(in.AccessType),
|
|
TimeExpires: &common.SDKTime{Time: time.Now().Add(time.Duration(in.ExpiresHours) * time.Hour)},
|
|
}
|
|
if in.ObjectName != "" {
|
|
details.ObjectName = &in.ObjectName
|
|
}
|
|
// 桶级 PAR(AnyObject*)必须显式桶列举动作,缺省会被 OCI 拒绝;一律放开列举便于收件人浏览
|
|
if strings.HasPrefix(in.AccessType, "AnyObject") {
|
|
details.BucketListingAction = objectstorage.PreauthenticatedRequestBucketListingActionListobjects
|
|
}
|
|
resp, err := oc.CreatePreauthenticatedRequest(ctx, objectstorage.CreatePreauthenticatedRequestRequest{
|
|
NamespaceName: &ns, BucketName: &bucket, CreatePreauthenticatedRequestDetails: details,
|
|
})
|
|
if err != nil {
|
|
return PAR{}, fmt.Errorf("create par: %w", err)
|
|
}
|
|
return toPAR(resp.PreauthenticatedRequest, oc.Endpoint()), nil
|
|
}
|
|
|
|
func toPAR(p objectstorage.PreauthenticatedRequest, endpoint string) PAR {
|
|
out := PAR{
|
|
ID: deref(p.Id),
|
|
Name: deref(p.Name),
|
|
ObjectName: deref(p.ObjectName),
|
|
AccessType: string(p.AccessType),
|
|
}
|
|
if p.AccessUri != nil {
|
|
out.FullURL = strings.TrimSuffix(endpoint, "/") + *p.AccessUri
|
|
}
|
|
if p.TimeExpires != nil {
|
|
out.TimeExpires = &p.TimeExpires.Time
|
|
}
|
|
if p.TimeCreated != nil {
|
|
out.TimeCreated = &p.TimeCreated.Time
|
|
}
|
|
return out
|
|
}
|
|
|
|
// ListPARsPage 实现 Client:单页列出桶内预签名请求(摘要不含 URL),
|
|
// 返回下一页游标,空串表示已到末页;limit ≤0 时用 OCI 默认页大小。
|
|
func (c *RealClient) ListPARsPage(ctx context.Context, cred Credentials, region, bucket, page string, limit int) ([]PAR, string, error) {
|
|
oc, err := c.osClient(cred, region)
|
|
if err != nil {
|
|
return nil, "", err
|
|
}
|
|
ns, err := c.GetObjectStorageNamespace(ctx, cred, region)
|
|
if err != nil {
|
|
return nil, "", err
|
|
}
|
|
req := objectstorage.ListPreauthenticatedRequestsRequest{NamespaceName: &ns, BucketName: &bucket}
|
|
if limit > 0 {
|
|
req.Limit = common.Int(limit)
|
|
}
|
|
if page != "" {
|
|
req.Page = &page
|
|
}
|
|
resp, err := oc.ListPreauthenticatedRequests(ctx, req)
|
|
if err != nil {
|
|
return nil, "", fmt.Errorf("list pars: %w", err)
|
|
}
|
|
out := make([]PAR, 0, len(resp.Items))
|
|
for _, it := range resp.Items {
|
|
out = append(out, summaryPAR(it))
|
|
}
|
|
return out, deref(resp.OpcNextPage), nil
|
|
}
|
|
|
|
// ListPARs 实现 Client:翻页列出桶内全部有效预签名请求(清空/全部删除用)。
|
|
func (c *RealClient) ListPARs(ctx context.Context, cred Credentials, region, bucket string) ([]PAR, error) {
|
|
out := make([]PAR, 0, 16)
|
|
page := ""
|
|
for {
|
|
items, next, err := c.ListPARsPage(ctx, cred, region, bucket, page, 1000)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, items...)
|
|
if next == "" {
|
|
return out, nil
|
|
}
|
|
page = next
|
|
}
|
|
}
|
|
|
|
func summaryPAR(p objectstorage.PreauthenticatedRequestSummary) PAR {
|
|
out := PAR{
|
|
ID: deref(p.Id),
|
|
Name: deref(p.Name),
|
|
ObjectName: deref(p.ObjectName),
|
|
AccessType: string(p.AccessType),
|
|
}
|
|
if p.TimeExpires != nil {
|
|
out.TimeExpires = &p.TimeExpires.Time
|
|
}
|
|
if p.TimeCreated != nil {
|
|
out.TimeCreated = &p.TimeCreated.Time
|
|
}
|
|
return out
|
|
}
|
|
|
|
// DeletePAR 实现 Client:撤销预签名请求,链接立即失效。
|
|
func (c *RealClient) DeletePAR(ctx context.Context, cred Credentials, region, bucket, parID string) error {
|
|
oc, err := c.osClient(cred, region)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
ns, err := c.GetObjectStorageNamespace(ctx, cred, region)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if _, err := oc.DeletePreauthenticatedRequest(ctx, objectstorage.DeletePreauthenticatedRequestRequest{
|
|
NamespaceName: &ns, BucketName: &bucket, ParId: &parID,
|
|
RequestMetadata: common.RequestMetadata{RetryPolicy: &bulkRetryPolicy},
|
|
}); err != nil {
|
|
return fmt.Errorf("delete par: %w", err)
|
|
}
|
|
return nil
|
|
}
|