feat(admin): 运行时策略 - 配置对象存储策略与默认并发
This commit is contained in:
@@ -324,6 +324,7 @@ func (s *Service) openResourceRange(userID string, resource *model.Resource, ran
|
||||
setting.Provider = firstNonEmpty(resource.Provider, setting.Provider)
|
||||
setting.Endpoint = firstNonEmpty(resource.Endpoint, setting.Endpoint)
|
||||
setting.Bucket = firstNonEmpty(resource.Bucket, setting.Bucket)
|
||||
setting = s.storageSettingWithRuntimePolicy(setting)
|
||||
stream, err := getOriginOSSObjectRange(setting, resource.ObjectKey, normalizeSingleByteRange(rangeHeader))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -445,7 +446,7 @@ func (s *Service) storeResourceObject(resource *model.Resource, fileName string,
|
||||
if settingErr != nil {
|
||||
return "", settingErr
|
||||
}
|
||||
etag, err := putOSSObject(setting, resource.ObjectKey, resource.MimeType, resource.Size, body)
|
||||
etag, err := putOSSObject(s.storageSettingWithRuntimePolicy(setting), resource.ObjectKey, resource.MimeType, resource.Size, body)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
@@ -1,14 +1,33 @@
|
||||
package app
|
||||
|
||||
import "infinite-canvas/backend/internal/storage"
|
||||
import (
|
||||
"time"
|
||||
|
||||
"infinite-canvas/backend/internal/storage"
|
||||
)
|
||||
|
||||
type ossSettingValue = storage.Settings
|
||||
type ossProviderCredentials = storage.Credentials
|
||||
type ossObjectStream = storage.ObjectStream
|
||||
|
||||
func (s *Service) storageSettingWithRuntimePolicy(setting ossSettingValue) ossSettingValue {
|
||||
setting = normalizeOSSSetting(setting)
|
||||
policy, err := s.RuntimePolicy()
|
||||
if err != nil {
|
||||
return setting
|
||||
}
|
||||
setting.Runtime = storage.RuntimePolicy{
|
||||
TransferTimeout: time.Duration(policy.Storage.TransferTimeoutSeconds) * time.Second,
|
||||
AccessURLTTL: time.Duration(policy.Storage.AccessURLTTLSeconds) * time.Second,
|
||||
ProviderAccessURLTTL: time.Duration(policy.Storage.ProviderAccessURLTTLSeconds) * time.Second,
|
||||
MaxBufferedUploadSize: policy.Storage.NonSeekableBufferMB * 1024 * 1024,
|
||||
ErrorBodyLimit: policy.Storage.ErrorBodyKB * 1024,
|
||||
}
|
||||
return normalizeOSSSetting(setting)
|
||||
}
|
||||
|
||||
var (
|
||||
putOSSObject = storage.PutOSSObject
|
||||
putAliyunOSSObject = storage.PutAliyunOSSObject
|
||||
getOSSObjectRange = storage.GetOSSObjectRange
|
||||
getAliyunOSSObjectRange = storage.GetAliyunOSSObjectRange
|
||||
signedOSSObjectURL = storage.SignedOSSObjectURL
|
||||
|
||||
@@ -81,7 +81,7 @@ func (s *Service) testOSSSetting(scope string, ownerID string, actorID string, r
|
||||
}()
|
||||
|
||||
testKey := path.Join(value.PathPrefix, ".yingce-tests", scope, newID())
|
||||
if err := verifyOSSConnection(value, testKey); err != nil {
|
||||
if err := verifyOSSConnection(s.storageSettingWithRuntimePolicy(value), testKey); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
|
||||
@@ -281,7 +281,7 @@ func (s *Service) storeTaskMediaObject(resource *model.Resource, _ string, body
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
return putOSSObject(setting, resource.ObjectKey, resource.MimeType, resource.Size, body)
|
||||
return putOSSObject(s.storageSettingWithRuntimePolicy(setting), resource.ObjectKey, resource.MimeType, resource.Size, body)
|
||||
}
|
||||
|
||||
// Cleanup is best effort only after durable completion. Expired leftovers are
|
||||
|
||||
@@ -24,7 +24,7 @@ type localRateEntry struct {
|
||||
const (
|
||||
MinChannelConcurrencyLimit = 1
|
||||
MaxChannelConcurrencyLimit = maxRuntimeConcurrency
|
||||
defaultChannelConcurrencyValue = 3
|
||||
defaultChannelConcurrencyValue = 8
|
||||
)
|
||||
|
||||
type channelSlotError struct {
|
||||
@@ -381,13 +381,17 @@ func envInt(key string, fallback int) int {
|
||||
return value
|
||||
}
|
||||
|
||||
func defaultWorkerConcurrencyLimit() int {
|
||||
return envInt("CANVAS_WORKER_CONCURRENCY", defaultWorkerConcurrency)
|
||||
}
|
||||
|
||||
func defaultChannelConcurrencyLimit() int {
|
||||
return effectiveChannelConcurrencyLimit(envInt("CANVAS_CHANNEL_CONCURRENCY", defaultChannelConcurrencyValue))
|
||||
return envInt("CANVAS_CHANNEL_CONCURRENCY", defaultChannelConcurrencyValue)
|
||||
}
|
||||
|
||||
func effectiveChannelConcurrencyLimit(configured int) int {
|
||||
if configured < MinChannelConcurrencyLimit || configured > MaxChannelConcurrencyLimit {
|
||||
return defaultChannelConcurrencyValue
|
||||
return defaultChannelConcurrencyLimit()
|
||||
}
|
||||
return configured
|
||||
}
|
||||
|
||||
@@ -8,7 +8,7 @@ import (
|
||||
"infinite-canvas/backend/internal/repository"
|
||||
)
|
||||
|
||||
const TaskWorkerConcurrency = 3
|
||||
const TaskWorkerConcurrency = 8
|
||||
|
||||
// Host 由组合根注入,避免 platform → app 回环。
|
||||
type Host interface {
|
||||
|
||||
@@ -15,13 +15,15 @@ import (
|
||||
const runtimePolicySettingKey = "runtime_policy"
|
||||
|
||||
const (
|
||||
maxRuntimeUploadMB int64 = 999
|
||||
maxRuntimeStorageGB int64 = 999
|
||||
maxRuntimeDataMB int64 = 999_999
|
||||
maxRuntimeCount int64 = 999_999_999
|
||||
maxRuntimeRate = 999_999
|
||||
maxRuntimeConcurrency = 999
|
||||
maxRuntimeTimeoutMinutes = 9_999
|
||||
maxRuntimeUploadMB int64 = 999
|
||||
maxRuntimeStorageGB int64 = 999
|
||||
maxRuntimeDataMB int64 = 999_999
|
||||
maxRuntimeCount int64 = 999_999_999
|
||||
maxRuntimeRate = 999_999
|
||||
maxRuntimeConcurrency = 999
|
||||
maxRuntimeTimeoutMinutes = 9_999
|
||||
defaultWorkerConcurrency = 8
|
||||
defaultChannelConcurrency = 8
|
||||
)
|
||||
|
||||
// 画布 Agent 单步的输出上限与执行时限。上限是"每次模型调用"的边界(思考 + 正文 + 工具参数),
|
||||
@@ -65,6 +67,14 @@ type RuntimeResourcePolicy struct {
|
||||
RecycleBinRetentionDays int `json:"recycleBinRetentionDays"`
|
||||
}
|
||||
|
||||
type RuntimeStoragePolicy struct {
|
||||
TransferTimeoutSeconds int `json:"transferTimeoutSeconds"`
|
||||
AccessURLTTLSeconds int `json:"accessURLTTLSeconds"`
|
||||
ProviderAccessURLTTLSeconds int `json:"providerAccessURLTTLSeconds"`
|
||||
NonSeekableBufferMB int64 `json:"nonSeekableBufferMB"`
|
||||
ErrorBodyKB int64 `json:"errorBodyKB"`
|
||||
}
|
||||
|
||||
type RuntimeTaskPolicy struct {
|
||||
WorkerConcurrency int `json:"workerConcurrency"`
|
||||
ChannelConcurrency int `json:"channelConcurrency"`
|
||||
@@ -106,6 +116,7 @@ type RuntimeRequestPolicy struct {
|
||||
|
||||
type RuntimePolicySetting struct {
|
||||
Resource RuntimeResourcePolicy `json:"resource"`
|
||||
Storage RuntimeStoragePolicy `json:"storage"`
|
||||
Task RuntimeTaskPolicy `json:"task"`
|
||||
Request RuntimeRequestPolicy `json:"request"`
|
||||
}
|
||||
@@ -143,8 +154,15 @@ func DefaultRuntimePolicy() RuntimePolicySetting {
|
||||
APICallLogCount: 100_000,
|
||||
RecycleBinRetentionDays: 30,
|
||||
},
|
||||
Storage: RuntimeStoragePolicy{
|
||||
TransferTimeoutSeconds: 120,
|
||||
AccessURLTTLSeconds: 300,
|
||||
ProviderAccessURLTTLSeconds: 14_400,
|
||||
NonSeekableBufferMB: 64,
|
||||
ErrorBodyKB: 1,
|
||||
},
|
||||
Task: RuntimeTaskPolicy{
|
||||
WorkerConcurrency: effectiveChannelConcurrencyLimit(envInt("CANVAS_WORKER_CONCURRENCY", TaskWorkerConcurrency)),
|
||||
WorkerConcurrency: defaultWorkerConcurrencyLimit(),
|
||||
ChannelConcurrency: defaultChannelConcurrencyLimit(),
|
||||
ActiveTaskLimit: 5,
|
||||
ImageTimeoutMinutes: 8,
|
||||
@@ -380,6 +398,22 @@ func validateRuntimePolicy(value RuntimePolicySetting) error {
|
||||
if resource.RecycleBinRetentionDays < 0 || resource.RecycleBinRetentionDays > 365 {
|
||||
return kernel.BadAuthRequest("回收站保留天数必须是 0-365 的整数 (0 表示不自动清理)")
|
||||
}
|
||||
storage := value.Storage
|
||||
if storage.TransferTimeoutSeconds < 10 || storage.TransferTimeoutSeconds > 3_600 {
|
||||
return kernel.BadAuthRequest("对象存储传输超时必须是 10-3600 秒的整数")
|
||||
}
|
||||
if storage.AccessURLTTLSeconds < 60 || storage.AccessURLTTLSeconds > 86_400 {
|
||||
return kernel.BadAuthRequest("对象存储访问地址有效期必须是 60-86400 秒的整数")
|
||||
}
|
||||
if storage.ProviderAccessURLTTLSeconds < 300 || storage.ProviderAccessURLTTLSeconds > 86_400 {
|
||||
return kernel.BadAuthRequest("模型输入资源地址有效期必须是 300-86400 秒的整数")
|
||||
}
|
||||
if storage.NonSeekableBufferMB < 1 || storage.NonSeekableBufferMB > maxRuntimeUploadMB {
|
||||
return kernel.BadAuthRequest(fmt.Sprintf("对象存储非可寻址上传缓冲必须是 1-%d MB 的整数", maxRuntimeUploadMB))
|
||||
}
|
||||
if storage.ErrorBodyKB < 1 || storage.ErrorBodyKB > 1024 {
|
||||
return kernel.BadAuthRequest("对象存储错误响应保留必须是 1-1024 KB 的整数")
|
||||
}
|
||||
task := value.Task
|
||||
for label, item := range map[string]int{
|
||||
"Worker 并发数": task.WorkerConcurrency, "全局渠道并发数": task.ChannelConcurrency, "活动任务上限": task.ActiveTaskLimit,
|
||||
|
||||
@@ -26,6 +26,30 @@ import (
|
||||
"time"
|
||||
)
|
||||
|
||||
func transferTimeout(setting Settings) time.Duration {
|
||||
setting = NormalizeSettings(setting)
|
||||
return setting.Runtime.TransferTimeout
|
||||
}
|
||||
|
||||
func accessURLTTL(setting Settings) time.Duration {
|
||||
setting = NormalizeSettings(setting)
|
||||
return setting.Runtime.AccessURLTTL
|
||||
}
|
||||
|
||||
func providerAccessURLTTL(setting Settings) time.Duration {
|
||||
setting = NormalizeSettings(setting)
|
||||
return setting.Runtime.ProviderAccessURLTTL
|
||||
}
|
||||
|
||||
func maxBufferedUploadSize(setting Settings) int64 {
|
||||
setting = NormalizeSettings(setting)
|
||||
return setting.Runtime.MaxBufferedUploadSize
|
||||
}
|
||||
|
||||
func errorBodyLimit(setting Settings) int64 {
|
||||
setting = NormalizeSettings(setting)
|
||||
return setting.Runtime.ErrorBodyLimit
|
||||
}
|
||||
func PutOSSObject(setting Settings, objectKey string, mimeType string, size int64, body io.Reader) (string, error) {
|
||||
setting = NormalizeSettings(setting)
|
||||
if setting.Provider == tencentCOSProvider {
|
||||
@@ -52,13 +76,13 @@ func PutAliyunOSSObject(setting Settings, objectKey string, mimeType string, siz
|
||||
if size > 0 {
|
||||
req.ContentLength = size
|
||||
}
|
||||
resp, err := outbound.OutboundHTTPClient(2 * time.Minute).Do(req)
|
||||
resp, err := outbound.OutboundHTTPClient(transferTimeout(setting)).Do(req)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if resp.StatusCode < 200 || resp.StatusCode >= 300 {
|
||||
detail, _ := io.ReadAll(io.LimitReader(resp.Body, 512))
|
||||
detail, _ := io.ReadAll(io.LimitReader(resp.Body, errorBodyLimit(setting)))
|
||||
return "", fmt.Errorf("OSS 上传失败:%s %s", resp.Status, strings.TrimSpace(string(detail)))
|
||||
}
|
||||
return strings.Trim(resp.Header.Get("ETag"), `"`), nil
|
||||
@@ -106,7 +130,7 @@ func getCDNObjectRange(setting Settings, objectKey string, rangeHeader string) (
|
||||
if setting.Delivery.CDNAuthMode == "" {
|
||||
urlValue, err = OssCDNObjectURL(setting.CDNBaseURL, objectKey)
|
||||
} else {
|
||||
urlValue, err = SignCDNURL(setting, objectKey, time.Now().Add(resourceAccessURLTTL))
|
||||
urlValue, err = SignCDNURL(setting, objectKey, time.Now().Add(accessURLTTL(setting)))
|
||||
}
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -119,19 +143,20 @@ func getCDNObjectRange(setting Settings, objectKey string, rangeHeader string) (
|
||||
req.Header.Set("Range", rangeHeader)
|
||||
}
|
||||
outbound.ApplyDefaultOutboundHeaders(req)
|
||||
resp, err := outbound.OutboundHTTPClient(2 * time.Minute).Do(req)
|
||||
resp, err := outbound.OutboundHTTPClient(transferTimeout(setting)).Do(req)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("CDN 读取失败:%w", err)
|
||||
}
|
||||
if (resp.StatusCode < 200 || resp.StatusCode >= 300) && resp.StatusCode != http.StatusRequestedRangeNotSatisfiable {
|
||||
defer resp.Body.Close()
|
||||
detail, _ := io.ReadAll(io.LimitReader(resp.Body, 512))
|
||||
detail, _ := io.ReadAll(io.LimitReader(resp.Body, errorBodyLimit(setting)))
|
||||
return nil, fmt.Errorf("CDN 读取失败:%s %s", resp.Status, strings.TrimSpace(string(detail)))
|
||||
}
|
||||
return &ObjectStream{Body: resp.Body, StatusCode: resp.StatusCode, ContentLength: resp.ContentLength, ContentRange: resp.Header.Get("Content-Range"), AcceptRanges: kernel.FirstNonEmpty(resp.Header.Get("Accept-Ranges"), "bytes")}, nil
|
||||
}
|
||||
|
||||
func GetAliyunOSSObjectRange(setting Settings, objectKey string, rangeHeader string) (*ObjectStream, error) {
|
||||
setting = NormalizeSettings(setting)
|
||||
req, err := NewOSSRequest(http.MethodGet, setting, objectKey, "", nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
@@ -139,13 +164,13 @@ func GetAliyunOSSObjectRange(setting Settings, objectKey string, rangeHeader str
|
||||
if rangeHeader != "" {
|
||||
req.Header.Set("Range", rangeHeader)
|
||||
}
|
||||
resp, err := outbound.OutboundHTTPClient(2 * time.Minute).Do(req)
|
||||
resp, err := outbound.OutboundHTTPClient(transferTimeout(setting)).Do(req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if (resp.StatusCode < 200 || resp.StatusCode >= 300) && resp.StatusCode != http.StatusRequestedRangeNotSatisfiable {
|
||||
defer resp.Body.Close()
|
||||
detail, _ := io.ReadAll(io.LimitReader(resp.Body, 512))
|
||||
detail, _ := io.ReadAll(io.LimitReader(resp.Body, errorBodyLimit(setting)))
|
||||
return nil, fmt.Errorf("OSS 读取失败:%s %s", resp.Status, strings.TrimSpace(string(detail)))
|
||||
}
|
||||
return &ObjectStream{Body: resp.Body, StatusCode: resp.StatusCode, ContentLength: resp.ContentLength, ContentRange: resp.Header.Get("Content-Range"), AcceptRanges: kernel.FirstNonEmpty(resp.Header.Get("Accept-Ranges"), "bytes")}, nil
|
||||
@@ -260,7 +285,8 @@ func signedAliyunOSSObjectURL(setting Settings, objectKey string, expiresAt time
|
||||
}
|
||||
|
||||
func PutCOSObject(setting Settings, objectKey string, mimeType string, size int64, body io.Reader) (string, error) {
|
||||
client, err := NewCOSClient(setting, 2*time.Minute)
|
||||
setting = NormalizeSettings(setting)
|
||||
client, err := NewCOSClient(setting, transferTimeout(setting))
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
@@ -276,6 +302,7 @@ func PutCOSObject(setting Settings, objectKey string, mimeType string, size int6
|
||||
}
|
||||
|
||||
func PutQiniuObject(setting Settings, objectKey string, mimeType string, size int64, body io.Reader) (string, error) {
|
||||
setting = NormalizeSettings(setting)
|
||||
if setting.AccessKeyID == "" || setting.AccessKeySecret == "" {
|
||||
return "", errors.New("七牛云 Kodo 访问密钥不可用")
|
||||
}
|
||||
@@ -300,7 +327,8 @@ func PutQiniuObject(setting Settings, objectKey string, mimeType string, size in
|
||||
}
|
||||
|
||||
func GetCOSObjectRange(setting Settings, objectKey string, rangeHeader string) (*ObjectStream, error) {
|
||||
client, err := NewCOSClient(setting, 2*time.Minute)
|
||||
setting = NormalizeSettings(setting)
|
||||
client, err := NewCOSClient(setting, transferTimeout(setting))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -316,7 +344,8 @@ func GetCOSObjectRange(setting Settings, objectKey string, rangeHeader string) (
|
||||
}
|
||||
|
||||
func GetQiniuObjectRange(setting Settings, objectKey string, rangeHeader string) (*ObjectStream, error) {
|
||||
signedURL, err := SignedQiniuS3ObjectURL(setting, objectKey, time.Now().Add(resourceAccessURLTTL))
|
||||
setting = NormalizeSettings(setting)
|
||||
signedURL, err := SignedQiniuS3ObjectURL(setting, objectKey, time.Now().Add(accessURLTTL(setting)))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -328,13 +357,13 @@ func GetQiniuObjectRange(setting Settings, objectKey string, rangeHeader string)
|
||||
req.Header.Set("Range", rangeHeader)
|
||||
}
|
||||
outbound.ApplyDefaultOutboundHeaders(req)
|
||||
resp, err := outbound.OutboundHTTPClient(2 * time.Minute).Do(req)
|
||||
resp, err := outbound.OutboundHTTPClient(transferTimeout(setting)).Do(req)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("七牛云 Kodo 读取失败:%w", err)
|
||||
}
|
||||
if (resp.StatusCode < 200 || resp.StatusCode >= 300) && resp.StatusCode != http.StatusRequestedRangeNotSatisfiable {
|
||||
defer resp.Body.Close()
|
||||
detail, _ := io.ReadAll(io.LimitReader(resp.Body, 512))
|
||||
detail, _ := io.ReadAll(io.LimitReader(resp.Body, errorBodyLimit(setting)))
|
||||
return nil, fmt.Errorf("七牛云 Kodo 读取失败:%s %s", resp.Status, strings.TrimSpace(string(detail)))
|
||||
}
|
||||
return &ObjectStream{Body: resp.Body, StatusCode: resp.StatusCode, ContentLength: resp.ContentLength, ContentRange: resp.Header.Get("Content-Range"), AcceptRanges: kernel.FirstNonEmpty(resp.Header.Get("Accept-Ranges"), "bytes")}, nil
|
||||
@@ -349,6 +378,7 @@ func SignedCOSObjectDownloadURL(setting Settings, objectKey string, expiresAt ti
|
||||
}
|
||||
|
||||
func signedCOSObjectURL(setting Settings, objectKey string, expiresAt time.Time, disposition string) (string, error) {
|
||||
setting = NormalizeSettings(setting)
|
||||
if strings.TrimSpace(setting.AccessKeyID) == "" || strings.TrimSpace(setting.AccessKeySecret) == "" {
|
||||
return "", errors.New("COS 访问密钥不可用")
|
||||
}
|
||||
@@ -360,7 +390,7 @@ func signedCOSObjectURL(setting Settings, objectKey string, expiresAt time.Time,
|
||||
if expires <= 0 {
|
||||
return "", errors.New("COS 签名有效期必须晚于当前时间")
|
||||
}
|
||||
client, err := NewCOSClient(setting, 2*time.Minute)
|
||||
client, err := NewCOSClient(setting, transferTimeout(setting))
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
@@ -405,6 +435,7 @@ func SignedQiniuS3ObjectDownloadURL(setting Settings, objectKey string, expiresA
|
||||
}
|
||||
|
||||
func signedQiniuS3ObjectURL(setting Settings, objectKey string, expiresAt time.Time, disposition string) (string, error) {
|
||||
setting = NormalizeSettings(setting)
|
||||
region := QiniuS3Region(setting)
|
||||
if region == "" {
|
||||
return "", errors.New("七牛云 Kodo S3 Region 不可用")
|
||||
@@ -612,24 +643,26 @@ func EscapeObjectKey(key string) string {
|
||||
}
|
||||
|
||||
func DeleteAliyunOSSObject(setting Settings, objectKey string) error {
|
||||
setting = NormalizeSettings(setting)
|
||||
req, err := NewOSSRequest(http.MethodDelete, setting, objectKey, "", nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
resp, err := outbound.OutboundHTTPClient(2 * time.Minute).Do(req)
|
||||
resp, err := outbound.OutboundHTTPClient(transferTimeout(setting)).Do(req)
|
||||
if err != nil {
|
||||
return fmt.Errorf("删除阿里云 OSS 对象失败:%w", err)
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
if (resp.StatusCode < 200 || resp.StatusCode >= 300) && resp.StatusCode != http.StatusNotFound {
|
||||
detail, _ := io.ReadAll(io.LimitReader(resp.Body, 512))
|
||||
detail, _ := io.ReadAll(io.LimitReader(resp.Body, errorBodyLimit(setting)))
|
||||
return fmt.Errorf("删除阿里云 OSS 对象失败:%s %s", resp.Status, strings.TrimSpace(string(detail)))
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func DeleteTencentCOSObject(setting Settings, objectKey string) error {
|
||||
client, err := NewCOSClient(setting, 2*time.Minute)
|
||||
setting = NormalizeSettings(setting)
|
||||
client, err := NewCOSClient(setting, transferTimeout(setting))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
@@ -647,6 +680,7 @@ func DeleteTencentCOSObject(setting Settings, objectKey string) error {
|
||||
}
|
||||
|
||||
func DeleteQiniuObject(setting Settings, objectKey string) error {
|
||||
setting = NormalizeSettings(setting)
|
||||
if setting.AccessKeyID == "" || setting.AccessKeySecret == "" {
|
||||
return errors.New("七牛云 Kodo 访问密钥不可用")
|
||||
}
|
||||
|
||||
@@ -76,7 +76,8 @@ func NewS3Client(setting Settings, timeout time.Duration) (*awss3.S3, error) {
|
||||
}
|
||||
|
||||
func PutS3Object(setting Settings, objectKey string, mimeType string, size int64, body io.Reader) (string, error) {
|
||||
client, err := NewS3Client(setting, 2*time.Minute)
|
||||
setting = NormalizeSettings(setting)
|
||||
client, err := NewS3Client(setting, transferTimeout(setting))
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
@@ -84,9 +85,9 @@ func PutS3Object(setting Settings, objectKey string, mimeType string, size int64
|
||||
if reader, ok := body.(io.ReadSeeker); ok {
|
||||
seekable = reader
|
||||
} else {
|
||||
limit := size + 1
|
||||
if limit <= 0 {
|
||||
limit = 64 << 20
|
||||
limit := maxBufferedUploadSize(setting) + 1
|
||||
if size >= 0 && size < limit {
|
||||
limit = size + 1
|
||||
}
|
||||
data, readErr := io.ReadAll(io.LimitReader(body, limit))
|
||||
if readErr != nil {
|
||||
@@ -112,7 +113,8 @@ func PutS3Object(setting Settings, objectKey string, mimeType string, size int64
|
||||
}
|
||||
|
||||
func GetS3ObjectRange(setting Settings, objectKey string, rangeHeader string) (*ObjectStream, error) {
|
||||
client, err := NewS3Client(setting, 2*time.Minute)
|
||||
setting = NormalizeSettings(setting)
|
||||
client, err := NewS3Client(setting, transferTimeout(setting))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
@@ -143,7 +145,8 @@ func SignedS3ObjectDownloadURL(setting Settings, objectKey string, expiresAt tim
|
||||
}
|
||||
|
||||
func signedS3ObjectURL(setting Settings, objectKey string, expiresAt time.Time, disposition string) (string, error) {
|
||||
client, err := NewS3Client(setting, 2*time.Minute)
|
||||
setting = NormalizeSettings(setting)
|
||||
client, err := NewS3Client(setting, transferTimeout(setting))
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
@@ -164,7 +167,8 @@ func signedS3ObjectURL(setting Settings, objectKey string, expiresAt time.Time,
|
||||
}
|
||||
|
||||
func DeleteS3Object(setting Settings, objectKey string) error {
|
||||
client, err := NewS3Client(setting, 2*time.Minute)
|
||||
setting = NormalizeSettings(setting)
|
||||
client, err := NewS3Client(setting, transferTimeout(setting))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
@@ -31,10 +31,19 @@ type Settings struct {
|
||||
SessionToken string `json:"sessionToken"`
|
||||
StorageLocationID string `json:"storageLocationId"`
|
||||
AllowUserS3 bool `json:"allowUserS3"`
|
||||
Runtime RuntimePolicy `json:"-"`
|
||||
// 平台切换云厂商后仍需读取历史资源,因此仅归档非当前厂商的访问密钥。
|
||||
ArchivedCredentials map[string]Credentials `json:"archivedCredentials,omitempty"`
|
||||
}
|
||||
|
||||
type RuntimePolicy struct {
|
||||
TransferTimeout time.Duration
|
||||
AccessURLTTL time.Duration
|
||||
ProviderAccessURLTTL time.Duration
|
||||
MaxBufferedUploadSize int64
|
||||
ErrorBodyLimit int64
|
||||
}
|
||||
|
||||
type Credentials struct {
|
||||
AccessKeyID string `json:"accessKeyId"`
|
||||
AccessKeySecret string `json:"accessKeySecret"`
|
||||
@@ -80,6 +89,21 @@ func NormalizeSettings(value Settings) Settings {
|
||||
}
|
||||
value.SessionToken = strings.TrimSpace(value.SessionToken)
|
||||
value.StorageLocationID = strings.TrimSpace(value.StorageLocationID)
|
||||
if value.Runtime.TransferTimeout <= 0 {
|
||||
value.Runtime.TransferTimeout = 2 * time.Minute
|
||||
}
|
||||
if value.Runtime.AccessURLTTL <= 0 {
|
||||
value.Runtime.AccessURLTTL = resourceAccessURLTTL
|
||||
}
|
||||
if value.Runtime.ProviderAccessURLTTL <= 0 {
|
||||
value.Runtime.ProviderAccessURLTTL = 4 * time.Hour
|
||||
}
|
||||
if value.Runtime.MaxBufferedUploadSize <= 0 {
|
||||
value.Runtime.MaxBufferedUploadSize = 64 << 20
|
||||
}
|
||||
if value.Runtime.ErrorBodyLimit <= 0 {
|
||||
value.Runtime.ErrorBodyLimit = 1024
|
||||
}
|
||||
value.ArchivedCredentials = CloneCredentials(value.ArchivedCredentials)
|
||||
return value
|
||||
}
|
||||
|
||||
@@ -9,8 +9,8 @@ import { useAdminContext } from "../admin-context";
|
||||
import { AdminPageFrame } from "../components/admin-shell";
|
||||
import { AdminStatTile, AdminStatusBadge, SettingsSectionCard } from "../components/admin-ui";
|
||||
|
||||
type PolicyGroup = "resource" | "task" | "request";
|
||||
type RuntimePolicyDraft = Pick<RuntimePolicySetting, "resource" | "task" | "request">;
|
||||
type PolicyGroup = "resource" | "storage" | "task" | "request";
|
||||
type RuntimePolicyDraft = Pick<RuntimePolicySetting, "resource" | "storage" | "task" | "request">;
|
||||
type PolicyField = { group: PolicyGroup; name: string; label: string; extra: string; unit: string; min?: number; max: number };
|
||||
type PolicySectionDefinition = {
|
||||
id: string;
|
||||
@@ -36,6 +36,13 @@ const resourceFields: PolicyField[] = [
|
||||
{ group: "resource", name: "recycleBinRetentionDays", label: "回收站自动清理时间", extra: "素材移入回收站后自动彻底删除的天数,0 表示不自动清理。", unit: "天", min: 0, max: 365 },
|
||||
];
|
||||
|
||||
const storageFields: PolicyField[] = [
|
||||
{ group: "storage", name: "transferTimeoutSeconds", label: "对象存储传输超时", extra: "对象存储上传、下载、删除与连接测试的最长等待时间。", unit: "秒", min: 10, max: 3_600 },
|
||||
{ group: "storage", name: "accessURLTTLSeconds", label: "访问地址有效期", extra: "浏览器访问对象资源时使用的签名地址有效期。", unit: "秒", min: 60, max: 86_400 },
|
||||
{ group: "storage", name: "providerAccessURLTTLSeconds", label: "厂商访问地址有效期", extra: "模型或其他外部厂商读取对象资源时使用的地址有效期。", unit: "秒", min: 300, max: 86_400 },
|
||||
{ group: "storage", name: "nonSeekableBufferMB", label: "非可寻址上传缓冲", extra: "S3 兼容存储接收不可重复读取请求体时的最大内存缓冲。", unit: "MB", max: 999 },
|
||||
{ group: "storage", name: "errorBodyKB", label: "错误响应保留", extra: "对象存储错误响应中最多保留的诊断内容。", unit: "KB", max: 1_024 },
|
||||
];
|
||||
const concurrencyFields: PolicyField[] = [
|
||||
{ group: "task", name: "workerConcurrency", label: "Worker 并发", extra: "集群同时执行的后台任务数。", unit: "个", max: 999 },
|
||||
{ group: "task", name: "channelConcurrency", label: "全局渠道并发", extra: "渠道选择跟随系统时采用的并发上限。", unit: "个", max: 999 },
|
||||
@@ -99,6 +106,15 @@ const relayFields: PolicyField[] = [
|
||||
|
||||
const policySections: PolicySectionDefinition[] = [
|
||||
{ id: "policy-resource", icon: <Database className="size-4" aria-hidden="true" />, title: "资源与账号配额", shortTitle: "资源配额", description: "上传、文件容量、结构化数据和历史记录上限。", fields: resourceFields },
|
||||
{
|
||||
id: "policy-storage",
|
||||
icon: <Database className="size-4" aria-hidden="true" />,
|
||||
title: "对象存储运行时",
|
||||
shortTitle: "对象存储",
|
||||
description: "对象存储传输、签名地址和上传缓冲策略。",
|
||||
fields: storageFields,
|
||||
status: <AdminStatusBadge label="保存后热更新" tone="info" />,
|
||||
},
|
||||
{
|
||||
id: "policy-concurrency",
|
||||
icon: <Gauge className="size-4" aria-hidden="true" />,
|
||||
@@ -526,7 +542,7 @@ function PolicySection({ icon, title, description, fields, status }: PolicySecti
|
||||
}
|
||||
|
||||
function toPolicyDraft(value: RuntimePolicyDraft): RuntimePolicyDraft {
|
||||
return { resource: { ...value.resource }, task: { ...value.task }, request: { ...value.request } };
|
||||
return { resource: { ...value.resource }, storage: { ...value.storage }, task: { ...value.task }, request: { ...value.request } };
|
||||
}
|
||||
|
||||
function readPolicyValue(value: RuntimePolicyDraft, field: PolicyField) {
|
||||
|
||||
@@ -7,7 +7,6 @@ import { http, apiBaseURL } from "@/services/api/request";
|
||||
import type { OSSConnectionTestInput, OSSConnectionTestResult, OSSProvider, S3Preset } from "@/lib/oss-settings";
|
||||
import type { VerificationPolicy } from "./verification";
|
||||
|
||||
|
||||
let authSessionRequest: Promise<AuthSessionPayload> | null = null;
|
||||
let authSessionCache: { payload: AuthSessionPayload; expiresAt: number } | null = null;
|
||||
|
||||
@@ -329,6 +328,14 @@ export type RuntimeResourcePolicy = {
|
||||
recycleBinRetentionDays?: number;
|
||||
};
|
||||
|
||||
export type RuntimeStoragePolicy = {
|
||||
transferTimeoutSeconds: number;
|
||||
accessURLTTLSeconds: number;
|
||||
providerAccessURLTTLSeconds: number;
|
||||
nonSeekableBufferMB: number;
|
||||
errorBodyKB: number;
|
||||
};
|
||||
|
||||
export type RuntimeTaskPolicy = {
|
||||
workerConcurrency: number;
|
||||
channelConcurrency: number;
|
||||
@@ -369,6 +376,7 @@ export type RuntimeRequestPolicy = {
|
||||
|
||||
export type RuntimePolicySetting = {
|
||||
resource: RuntimeResourcePolicy;
|
||||
storage: RuntimeStoragePolicy;
|
||||
task: RuntimeTaskPolicy;
|
||||
request: RuntimeRequestPolicy;
|
||||
configured?: boolean;
|
||||
@@ -378,7 +386,19 @@ export type RuntimePolicySetting = {
|
||||
};
|
||||
|
||||
export function getAuthSettings() {
|
||||
return http.get<VerificationPolicy & { firstUser: boolean; registrationEnabled: boolean; linuxdoEnabled: boolean; emailEnabled: boolean; emailCodeRequired: boolean; smsBindingAvailable: boolean; emailBindingAvailable: boolean; agreementTitle?: string; agreementContent?: string }>("/auth/settings");
|
||||
return http.get<
|
||||
VerificationPolicy & {
|
||||
firstUser: boolean;
|
||||
registrationEnabled: boolean;
|
||||
linuxdoEnabled: boolean;
|
||||
emailEnabled: boolean;
|
||||
emailCodeRequired: boolean;
|
||||
smsBindingAvailable: boolean;
|
||||
emailBindingAvailable: boolean;
|
||||
agreementTitle?: string;
|
||||
agreementContent?: string;
|
||||
}
|
||||
>("/auth/settings");
|
||||
}
|
||||
|
||||
export function linuxDOLoginURL(next: string, acceptedTerms?: boolean) {
|
||||
@@ -391,7 +411,8 @@ export function getAuthSession() {
|
||||
const now = Date.now();
|
||||
if (authSessionCache && authSessionCache.expiresAt > now) return Promise.resolve(authSessionCache.payload);
|
||||
if (authSessionRequest) return authSessionRequest;
|
||||
authSessionRequest = http.get<AuthSessionPayload>("/auth/session")
|
||||
authSessionRequest = http
|
||||
.get<AuthSessionPayload>("/auth/session")
|
||||
.then((payload) => {
|
||||
authSessionCache = { payload, expiresAt: Date.now() + 5_000 };
|
||||
return payload;
|
||||
@@ -414,7 +435,9 @@ export function getAdminFeatureAvailability() {
|
||||
return http.get<{ features: FeatureAvailability }>("/admin/settings/features");
|
||||
}
|
||||
|
||||
export function updateAdminFeatureAvailability(features: Partial<Pick<FeatureAvailability, "welcomeEnabled" | "shortDramaEnabled" | "taskCenterEnabled" | "creditsEnabled" | "customChannelsEnabled" | "frontendModelsEnabled" | "pluginCenterEnabled" | "systemPluginsVisibleToUsers">>) {
|
||||
export function updateAdminFeatureAvailability(
|
||||
features: Partial<Pick<FeatureAvailability, "welcomeEnabled" | "shortDramaEnabled" | "taskCenterEnabled" | "creditsEnabled" | "customChannelsEnabled" | "frontendModelsEnabled" | "pluginCenterEnabled" | "systemPluginsVisibleToUsers">>,
|
||||
) {
|
||||
return http.patch<{ features: FeatureAvailability }>("/admin/settings/features", features);
|
||||
}
|
||||
|
||||
@@ -567,7 +590,7 @@ export function getAdminSelfUseRuntimePolicy() {
|
||||
return http.get<{ setting: RuntimePolicySetting }>("/admin/settings/runtime-policy/self-use");
|
||||
}
|
||||
|
||||
export function updateAdminRuntimePolicySetting(input: Pick<RuntimePolicySetting, "resource" | "task" | "request">) {
|
||||
export function updateAdminRuntimePolicySetting(input: Pick<RuntimePolicySetting, "resource" | "storage" | "task" | "request">) {
|
||||
return http.put<{ setting: RuntimePolicySetting }>("/admin/settings/runtime-policy", input);
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user