mirror of
https://github.com/DullJZ/s3-balance.git
synced 2026-09-06 16:27:38 +08:00
Fix s3balance config update
This commit is contained in:
@@ -108,6 +108,9 @@ func main() {
|
|||||||
log.Printf("Failed to update load balancer strategy: %v", err)
|
log.Printf("Failed to update load balancer strategy: %v", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// 更新S3 API设置
|
||||||
|
s3Handler.UpdateS3APIConfig(&newConfig.S3API)
|
||||||
|
|
||||||
log.Println("Components updated successfully")
|
log.Println("Components updated successfully")
|
||||||
})
|
})
|
||||||
|
|
||||||
@@ -254,7 +257,7 @@ func startSessionCleaner(ctx context.Context, storageService *storage.Service) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// cleanupS3MultipartUploads 清理S3存储桶中过期的分片上传
|
// cleanupS3MultipartUploads 清理S3存储桶中过期的分片上传
|
||||||
func cleanupS3MultipartUploads(ctx context.Context, storageService *storage.Service) {
|
func cleanupS3MultipartUploads(_ context.Context, storageService *storage.Service) {
|
||||||
// 获取所有过期的会话
|
// 获取所有过期的会话
|
||||||
sessions, err := storageService.GetPendingUploadSessions("", "", "", 0)
|
sessions, err := storageService.GetPendingUploadSessions("", "", "", 0)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
@@ -7,6 +7,7 @@ require (
|
|||||||
github.com/aws/aws-sdk-go-v2/config v1.31.1
|
github.com/aws/aws-sdk-go-v2/config v1.31.1
|
||||||
github.com/aws/aws-sdk-go-v2/credentials v1.18.5
|
github.com/aws/aws-sdk-go-v2/credentials v1.18.5
|
||||||
github.com/aws/aws-sdk-go-v2/service/s3 v1.87.0
|
github.com/aws/aws-sdk-go-v2/service/s3 v1.87.0
|
||||||
|
github.com/aws/smithy-go v1.22.5
|
||||||
github.com/fsnotify/fsnotify v1.9.0
|
github.com/fsnotify/fsnotify v1.9.0
|
||||||
github.com/gorilla/mux v1.8.1
|
github.com/gorilla/mux v1.8.1
|
||||||
github.com/prometheus/client_golang v1.23.2
|
github.com/prometheus/client_golang v1.23.2
|
||||||
@@ -33,7 +34,6 @@ require (
|
|||||||
github.com/aws/aws-sdk-go-v2/service/sso v1.28.1 // indirect
|
github.com/aws/aws-sdk-go-v2/service/sso v1.28.1 // indirect
|
||||||
github.com/aws/aws-sdk-go-v2/service/ssooidc v1.33.1 // indirect
|
github.com/aws/aws-sdk-go-v2/service/ssooidc v1.33.1 // indirect
|
||||||
github.com/aws/aws-sdk-go-v2/service/sts v1.37.1 // indirect
|
github.com/aws/aws-sdk-go-v2/service/sts v1.37.1 // indirect
|
||||||
github.com/aws/smithy-go v1.22.5 // indirect
|
|
||||||
github.com/beorn7/perks v1.0.1 // indirect
|
github.com/beorn7/perks v1.0.1 // indirect
|
||||||
github.com/cespare/xxhash/v2 v2.3.0 // indirect
|
github.com/cespare/xxhash/v2 v2.3.0 // indirect
|
||||||
github.com/dustin/go-humanize v1.0.1 // indirect
|
github.com/dustin/go-humanize v1.0.1 // indirect
|
||||||
|
|||||||
@@ -84,7 +84,7 @@ func (h *S3Handler) handleGetObject(w http.ResponseWriter, r *http.Request, buck
|
|||||||
}
|
}
|
||||||
|
|
||||||
// 根据配置决定使用代理模式还是重定向模式
|
// 根据配置决定使用代理模式还是重定向模式
|
||||||
if h.proxyMode {
|
if h.proxyModeEnabled() {
|
||||||
// 代理模式:流式传输内容给客户端
|
// 代理模式:流式传输内容给客户端
|
||||||
resp, err := http.Get(downloadInfo.URL)
|
resp, err := http.Get(downloadInfo.URL)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
+67
-16
@@ -1,8 +1,11 @@
|
|||||||
package api
|
package api
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"sync/atomic"
|
||||||
|
|
||||||
"github.com/DullJZ/s3-balance/internal/balancer"
|
"github.com/DullJZ/s3-balance/internal/balancer"
|
||||||
"github.com/DullJZ/s3-balance/internal/bucket"
|
"github.com/DullJZ/s3-balance/internal/bucket"
|
||||||
|
"github.com/DullJZ/s3-balance/internal/config"
|
||||||
"github.com/DullJZ/s3-balance/internal/metrics"
|
"github.com/DullJZ/s3-balance/internal/metrics"
|
||||||
"github.com/DullJZ/s3-balance/internal/middleware"
|
"github.com/DullJZ/s3-balance/internal/middleware"
|
||||||
"github.com/DullJZ/s3-balance/internal/storage"
|
"github.com/DullJZ/s3-balance/internal/storage"
|
||||||
@@ -16,12 +19,16 @@ type S3Handler struct {
|
|||||||
balancer *balancer.Balancer
|
balancer *balancer.Balancer
|
||||||
presigner *presigner.Presigner
|
presigner *presigner.Presigner
|
||||||
storage *storage.Service
|
storage *storage.Service
|
||||||
accessKey string
|
|
||||||
secretKey string
|
|
||||||
metrics *metrics.Metrics
|
metrics *metrics.Metrics
|
||||||
proxyMode bool
|
settings atomic.Value
|
||||||
authRequired bool
|
}
|
||||||
virtualHost bool
|
|
||||||
|
type handlerSettings struct {
|
||||||
|
accessKey string
|
||||||
|
secretKey string
|
||||||
|
proxyMode bool
|
||||||
|
authRequired bool
|
||||||
|
virtualHost bool
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewS3Handler 创建新的S3兼容API处理器
|
// NewS3Handler 创建新的S3兼容API处理器
|
||||||
@@ -36,19 +43,27 @@ func NewS3Handler(
|
|||||||
proxyMode bool,
|
proxyMode bool,
|
||||||
authRequired bool,
|
authRequired bool,
|
||||||
virtualHost bool,
|
virtualHost bool,
|
||||||
|
|
||||||
) *S3Handler {
|
) *S3Handler {
|
||||||
return &S3Handler{
|
handler := &S3Handler{
|
||||||
bucketManager: bucketManager,
|
bucketManager: bucketManager,
|
||||||
balancer: balancer,
|
balancer: balancer,
|
||||||
presigner: presigner,
|
presigner: presigner,
|
||||||
storage: storage,
|
storage: storage,
|
||||||
accessKey: accessKey,
|
|
||||||
secretKey: secretKey,
|
|
||||||
metrics: metrics,
|
metrics: metrics,
|
||||||
proxyMode: proxyMode,
|
|
||||||
authRequired: authRequired,
|
|
||||||
virtualHost: virtualHost,
|
|
||||||
}
|
}
|
||||||
|
handler.initSettings(accessKey, secretKey, proxyMode, authRequired, virtualHost)
|
||||||
|
return handler
|
||||||
|
}
|
||||||
|
|
||||||
|
func (h *S3Handler) initSettings(accessKey, secretKey string, proxyMode, authRequired, virtualHost bool) {
|
||||||
|
h.settings.Store(handlerSettings{
|
||||||
|
accessKey: accessKey,
|
||||||
|
secretKey: secretKey,
|
||||||
|
proxyMode: proxyMode,
|
||||||
|
authRequired: authRequired,
|
||||||
|
virtualHost: virtualHost,
|
||||||
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
// RegisterS3Routes 注册S3兼容的路由
|
// RegisterS3Routes 注册S3兼容的路由
|
||||||
@@ -78,16 +93,52 @@ func (h *S3Handler) RegisterS3Routes(router *mux.Router) {
|
|||||||
|
|
||||||
// 添加中间件
|
// 添加中间件
|
||||||
router.Use(middleware.VirtualHost(middleware.VirtualHostConfig{
|
router.Use(middleware.VirtualHost(middleware.VirtualHostConfig{
|
||||||
Enabled: h.virtualHost,
|
Enabled: h.virtualHostEnabled,
|
||||||
BucketExists: func(name string) bool {
|
BucketExists: func(name string) bool {
|
||||||
_, ok := h.bucketManager.GetBucket(name)
|
_, ok := h.bucketManager.GetBucket(name)
|
||||||
return ok
|
return ok
|
||||||
},
|
},
|
||||||
}))
|
}))
|
||||||
router.Use(middleware.BasicAuth(middleware.AuthConfig{
|
router.Use(middleware.BasicAuth(middleware.AuthConfig{
|
||||||
Required: h.authRequired,
|
Required: h.authRequired,
|
||||||
AccessKey: h.accessKey,
|
Credentials: h.credentials,
|
||||||
SecretKey: h.secretKey,
|
OnError: h.sendS3Error,
|
||||||
OnError: h.sendS3Error,
|
|
||||||
}))
|
}))
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (h *S3Handler) loadSettings() handlerSettings {
|
||||||
|
if v := h.settings.Load(); v != nil {
|
||||||
|
return v.(handlerSettings)
|
||||||
|
}
|
||||||
|
return handlerSettings{}
|
||||||
|
}
|
||||||
|
|
||||||
|
func (h *S3Handler) virtualHostEnabled() bool {
|
||||||
|
return h.loadSettings().virtualHost
|
||||||
|
}
|
||||||
|
|
||||||
|
func (h *S3Handler) authRequired() bool {
|
||||||
|
return h.loadSettings().authRequired
|
||||||
|
}
|
||||||
|
|
||||||
|
func (h *S3Handler) credentials() (string, string) {
|
||||||
|
s := h.loadSettings()
|
||||||
|
return s.accessKey, s.secretKey
|
||||||
|
}
|
||||||
|
|
||||||
|
func (h *S3Handler) proxyModeEnabled() bool {
|
||||||
|
return h.loadSettings().proxyMode
|
||||||
|
}
|
||||||
|
|
||||||
|
func (h *S3Handler) UpdateS3APIConfig(cfg *config.S3APIConfig) {
|
||||||
|
if cfg == nil {
|
||||||
|
return
|
||||||
|
}
|
||||||
|
h.settings.Store(handlerSettings{
|
||||||
|
accessKey: cfg.AccessKey,
|
||||||
|
secretKey: cfg.SecretKey,
|
||||||
|
proxyMode: cfg.ProxyMode,
|
||||||
|
authRequired: cfg.AuthRequired,
|
||||||
|
virtualHost: cfg.VirtualHost,
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|||||||
@@ -8,17 +8,20 @@ import (
|
|||||||
|
|
||||||
// AuthConfig controls Basic Auth validation.
|
// AuthConfig controls Basic Auth validation.
|
||||||
type AuthConfig struct {
|
type AuthConfig struct {
|
||||||
Required bool
|
Required func() bool
|
||||||
AccessKey string
|
Credentials func() (string, string)
|
||||||
SecretKey string
|
OnError func(http.ResponseWriter, string, string, string)
|
||||||
OnError func(http.ResponseWriter, string, string, string)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// BasicAuth enforces static access/secret key authentication when Required is true.
|
// BasicAuth enforces static access/secret key authentication when required.
|
||||||
func BasicAuth(cfg AuthConfig) func(http.Handler) http.Handler {
|
func BasicAuth(cfg AuthConfig) func(http.Handler) http.Handler {
|
||||||
return func(next http.Handler) http.Handler {
|
return func(next http.Handler) http.Handler {
|
||||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
if !cfg.Required {
|
required := false
|
||||||
|
if cfg.Required != nil {
|
||||||
|
required = cfg.Required()
|
||||||
|
}
|
||||||
|
if !required {
|
||||||
next.ServeHTTP(w, r)
|
next.ServeHTTP(w, r)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
@@ -42,11 +45,16 @@ func BasicAuth(cfg AuthConfig) func(http.Handler) http.Handler {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
if parts[0] != cfg.AccessKey {
|
accessKey, secretKey := "", ""
|
||||||
|
if cfg.Credentials != nil {
|
||||||
|
accessKey, secretKey = cfg.Credentials()
|
||||||
|
}
|
||||||
|
|
||||||
|
if parts[0] != accessKey {
|
||||||
invokeOnError(w, cfg, "InvalidAccessKeyId", "The AWS Access Key Id you provided does not match the configured key.")
|
invokeOnError(w, cfg, "InvalidAccessKeyId", "The AWS Access Key Id you provided does not match the configured key.")
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
if parts[1] != cfg.SecretKey {
|
if parts[1] != secretKey {
|
||||||
invokeOnError(w, cfg, "SignatureDoesNotMatch", "The request signature we calculated does not match the signature you provided.")
|
invokeOnError(w, cfg, "SignatureDoesNotMatch", "The request signature we calculated does not match the signature you provided.")
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -8,7 +8,7 @@ import (
|
|||||||
|
|
||||||
// VirtualHostConfig controls host-style bucket resolution.
|
// VirtualHostConfig controls host-style bucket resolution.
|
||||||
type VirtualHostConfig struct {
|
type VirtualHostConfig struct {
|
||||||
Enabled bool
|
Enabled func() bool
|
||||||
BucketExists func(string) bool
|
BucketExists func(string) bool
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -16,7 +16,11 @@ type VirtualHostConfig struct {
|
|||||||
func VirtualHost(cfg VirtualHostConfig) func(http.Handler) http.Handler {
|
func VirtualHost(cfg VirtualHostConfig) func(http.Handler) http.Handler {
|
||||||
return func(next http.Handler) http.Handler {
|
return func(next http.Handler) http.Handler {
|
||||||
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
if !cfg.Enabled {
|
enabled := false
|
||||||
|
if cfg.Enabled != nil {
|
||||||
|
enabled = cfg.Enabled()
|
||||||
|
}
|
||||||
|
if !enabled {
|
||||||
next.ServeHTTP(w, r)
|
next.ServeHTTP(w, r)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user