Compare commits
1 Commits
ops-test-0
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
| b6ffe2fcea |
@@ -1,7 +1,13 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"context"
|
||||
"errors"
|
||||
"net/http"
|
||||
"os"
|
||||
"os/signal"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"git.apinb.com/bsm-sdk/core/infra"
|
||||
"git.apinb.com/bsm-sdk/core/middleware"
|
||||
@@ -16,12 +22,33 @@ import (
|
||||
const ServiceKey = "Files"
|
||||
|
||||
func main() {
|
||||
ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
|
||||
defer stop()
|
||||
|
||||
config.New(ServiceKey)
|
||||
impl.NewImpl()
|
||||
if err := models.InitData(); err != nil {
|
||||
if err := models.RequireSchemaVersion(impl.DBService, "files"); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
jobs.StartPendingUploadCleanup()
|
||||
if err := models.RequireStorageProvider(impl.DBService, impl.StorageService.Provider()); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
if err := jobs.StartPendingUploadCleanup(ctx); err != nil {
|
||||
if closeErr := impl.Close(); closeErr != nil {
|
||||
panic(errors.Join(err, closeErr))
|
||||
}
|
||||
panic(err)
|
||||
}
|
||||
if err := jobs.StartIntegrity(ctx); err != nil {
|
||||
stop()
|
||||
waitCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
waitErr := jobs.Wait(waitCtx)
|
||||
cancel()
|
||||
if closeErr := impl.Close(); closeErr != nil {
|
||||
panic(errors.Join(err, waitErr, closeErr))
|
||||
}
|
||||
panic(errors.Join(err, waitErr))
|
||||
}
|
||||
|
||||
app := gin.Default()
|
||||
middleware.Mode(app)
|
||||
@@ -30,7 +57,30 @@ func main() {
|
||||
app.HEAD("/", infra.Health)
|
||||
routers.Register(ServiceKey, app)
|
||||
|
||||
if err := app.Run(fmt.Sprintf(":%s", config.Spec.Port)); err != nil {
|
||||
serveErr := serve(ctx, app)
|
||||
stop()
|
||||
waitCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
waitErr := jobs.Wait(waitCtx)
|
||||
cancel()
|
||||
closeErr := impl.Close()
|
||||
if err := errors.Join(serveErr, waitErr, closeErr); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
}
|
||||
|
||||
func serve(ctx context.Context, app http.Handler) error {
|
||||
server := &http.Server{Addr: config.Spec.Addr, Handler: app, ReadHeaderTimeout: 10 * time.Second}
|
||||
errCh := make(chan error, 1)
|
||||
go func() { errCh <- server.ListenAndServe() }()
|
||||
select {
|
||||
case err := <-errCh:
|
||||
if errors.Is(err, http.ErrServerClosed) {
|
||||
return nil
|
||||
}
|
||||
return err
|
||||
case <-ctx.Done():
|
||||
shutdownCtx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
|
||||
defer cancel()
|
||||
return server.Shutdown(shutdownCtx)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,11 +1,12 @@
|
||||
Service: files
|
||||
Port: 12452
|
||||
BindIP: 0.0.0.0
|
||||
SecretKey: ops-files
|
||||
|
||||
Databases:
|
||||
Driver: postgres
|
||||
Source:
|
||||
- host=8.137.107.29 user=postgres password=Weidong2023~! dbname=ops_dev port=19432 sslmode=disable TimeZone=Asia/Shanghai
|
||||
- "${OPS_FILES_DB_DSN}"
|
||||
|
||||
Cache: redis://null:Weidong2023~!@8.137.107.29:19379/
|
||||
|
||||
@@ -19,17 +20,26 @@ ServiceClients:
|
||||
dc-control: ${DC_CONTROL_FILES_SECRET}
|
||||
visual: ${VISUAL_FILES_SECRET}
|
||||
|
||||
ObjectStorage:
|
||||
Provider: aliyun
|
||||
Endpoint: https://oss-cn-beijing.aliyuncs.com
|
||||
Region: cn-beijing
|
||||
Bucket: ops-app
|
||||
PublicBaseURL: https://ops-app.oss-cn-beijing.aliyuncs.com
|
||||
AccessKeyID: ${FILES_OSS_ACCESS_KEY_ID}
|
||||
AccessKeySecret: ${FILES_OSS_ACCESS_KEY_SECRET}
|
||||
PresignTTLSeconds: 600
|
||||
Storage:
|
||||
Provider: local
|
||||
ExternalBaseURL: http://127.0.0.1:12452/Files
|
||||
AccessTTLSeconds: 600
|
||||
SigningSecret: local-dev-files-signing-secret
|
||||
Local:
|
||||
RootPath: D:/work/ops/.data/files
|
||||
MinimumFreeSpaceMB: 1024
|
||||
Aliyun:
|
||||
Endpoint: ""
|
||||
Region: ""
|
||||
Bucket: ""
|
||||
AccessKeyID: ""
|
||||
AccessKeySecret: ""
|
||||
|
||||
Namespaces:
|
||||
device-config-backups:
|
||||
Prefix: device-config-backups
|
||||
MaxSizeMB: 1
|
||||
AllowedExtensions: [.txt]
|
||||
reports:
|
||||
Prefix: reports
|
||||
MaxSizeMB: 512
|
||||
@@ -45,3 +55,5 @@ Namespaces:
|
||||
|
||||
Cleanup:
|
||||
IntervalSeconds: 600
|
||||
IntegrityIntervalSeconds: 3600
|
||||
IntegrityBatchSize: 100
|
||||
|
||||
@@ -1,13 +1,14 @@
|
||||
Service: files
|
||||
Port: 12452
|
||||
SecretKey: ops-files
|
||||
BindIP: 127.0.0.1
|
||||
SecretKey: ${FILES_SERVICE_SECRET}
|
||||
|
||||
Databases:
|
||||
Driver: postgres
|
||||
Source:
|
||||
- host=8.137.107.29 user=postgres password=Weidong2023~! dbname=ops_dev port=19432 sslmode=disable TimeZone=Asia/Shanghai
|
||||
- "${OPS_FILES_DB_DSN}"
|
||||
|
||||
Cache: redis://null:Weidong2023~!@8.137.107.29:19379/
|
||||
Cache: ${OPS_CACHE_DSN}
|
||||
|
||||
MicroService:
|
||||
Enable: false
|
||||
@@ -19,17 +20,26 @@ ServiceClients:
|
||||
dc-control: ${DC_CONTROL_FILES_SECRET}
|
||||
visual: ${VISUAL_FILES_SECRET}
|
||||
|
||||
ObjectStorage:
|
||||
Provider: aliyun
|
||||
Endpoint: https://oss-cn-beijing.aliyuncs.com
|
||||
Region: cn-beijing
|
||||
Bucket: ops-app
|
||||
PublicBaseURL: https://ops-app.oss-cn-beijing.aliyuncs.com
|
||||
Storage:
|
||||
Provider: ${FILES_STORAGE_PROVIDER}
|
||||
ExternalBaseURL: ${FILES_PUBLIC_BASE_URL}
|
||||
AccessTTLSeconds: 600
|
||||
SigningSecret: ${FILES_STORAGE_SIGNING_SECRET}
|
||||
Local:
|
||||
RootPath: ${FILES_LOCAL_ROOT}
|
||||
MinimumFreeSpaceMB: 10240
|
||||
Aliyun:
|
||||
Endpoint: ${FILES_OSS_ENDPOINT}
|
||||
Region: ${FILES_OSS_REGION}
|
||||
Bucket: ${FILES_OSS_BUCKET}
|
||||
AccessKeyID: ${FILES_OSS_ACCESS_KEY_ID}
|
||||
AccessKeySecret: ${FILES_OSS_ACCESS_KEY_SECRET}
|
||||
PresignTTLSeconds: 600
|
||||
|
||||
Namespaces:
|
||||
device-config-backups:
|
||||
Prefix: device-config-backups
|
||||
MaxSizeMB: 1
|
||||
AllowedExtensions: [.txt]
|
||||
reports:
|
||||
Prefix: reports
|
||||
MaxSizeMB: 512
|
||||
@@ -45,3 +55,5 @@ Namespaces:
|
||||
|
||||
Cleanup:
|
||||
IntervalSeconds: 600
|
||||
IntegrityIntervalSeconds: 3600
|
||||
IntegrityBatchSize: 100
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
Service: files
|
||||
Port: 12452
|
||||
BindIP: 127.0.0.1
|
||||
SecretKey: ops-files
|
||||
|
||||
Databases:
|
||||
@@ -19,17 +20,26 @@ ServiceClients:
|
||||
dc-control: ${DC_CONTROL_FILES_SECRET}
|
||||
visual: ${VISUAL_FILES_SECRET}
|
||||
|
||||
ObjectStorage:
|
||||
Provider: aliyun
|
||||
Endpoint: https://oss-cn-beijing.aliyuncs.com
|
||||
Region: cn-beijing
|
||||
Bucket: ops-app
|
||||
PublicBaseURL: https://ops-app.oss-cn-beijing.aliyuncs.com
|
||||
AccessKeyID: ${FILES_OSS_ACCESS_KEY_ID}
|
||||
AccessKeySecret: ${FILES_OSS_ACCESS_KEY_SECRET}
|
||||
PresignTTLSeconds: 600
|
||||
Storage:
|
||||
Provider: local
|
||||
ExternalBaseURL: http://127.0.0.1:12452/Files
|
||||
AccessTTLSeconds: 600
|
||||
SigningSecret: local-test-files-signing-secret
|
||||
Local:
|
||||
RootPath: D:/work/ops/.data/files-test
|
||||
MinimumFreeSpaceMB: 1
|
||||
Aliyun:
|
||||
Endpoint: ""
|
||||
Region: ""
|
||||
Bucket: ""
|
||||
AccessKeyID: ""
|
||||
AccessKeySecret: ""
|
||||
|
||||
Namespaces:
|
||||
device-config-backups:
|
||||
Prefix: device-config-backups
|
||||
MaxSizeMB: 1
|
||||
AllowedExtensions: [.txt]
|
||||
reports:
|
||||
Prefix: reports
|
||||
MaxSizeMB: 512
|
||||
@@ -45,3 +55,5 @@ Namespaces:
|
||||
|
||||
Cleanup:
|
||||
IntervalSeconds: 600
|
||||
IntegrityIntervalSeconds: 3600
|
||||
IntegrityBatchSize: 100
|
||||
|
||||
@@ -13,11 +13,11 @@ Environment=BSM_RuntimeMode=prod
|
||||
Environment=RUN_MODE=prod
|
||||
Environment=BSM_Prefix=/data/app
|
||||
ExecStart=/data/app/ops-files
|
||||
ExecStartPost=/data/app/systemd/wait-http.sh ops-files http://127.0.0.1:12452/Files/v1/ping/hello 60
|
||||
ExecStartPost=/data/app/systemd/wait-http.sh ops-files http://127.0.0.1:12452/ready 60
|
||||
Restart=on-failure
|
||||
RestartSec=5s
|
||||
TimeoutStartSec=75s
|
||||
TimeoutStopSec=30s
|
||||
TimeoutStopSec=75s
|
||||
KillSignal=SIGTERM
|
||||
StandardOutput=append:/data/app/logs/files.log
|
||||
StandardError=inherit
|
||||
|
||||
4
go.mod
4
go.mod
@@ -23,6 +23,7 @@ require (
|
||||
github.com/gabriel-vasile/mimetype v1.4.11 // indirect
|
||||
github.com/gin-contrib/cors v1.7.6 // indirect
|
||||
github.com/gin-contrib/sse v1.1.0 // indirect
|
||||
github.com/go-ole/go-ole v1.2.6 // indirect
|
||||
github.com/go-playground/locales v0.14.1 // indirect
|
||||
github.com/go-playground/universal-translator v0.18.1 // indirect
|
||||
github.com/go-playground/validator/v10 v10.28.0 // indirect
|
||||
@@ -38,11 +39,13 @@ require (
|
||||
github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
|
||||
github.com/modern-go/reflect2 v1.0.2 // indirect
|
||||
github.com/pelletier/go-toml/v2 v2.2.4 // indirect
|
||||
github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect
|
||||
github.com/quic-go/qpack v0.5.1 // indirect
|
||||
github.com/quic-go/quic-go v0.55.0 // indirect
|
||||
github.com/skip2/go-qrcode v0.0.0-20200617195104-da1b6568686e // indirect
|
||||
github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
|
||||
github.com/ugorji/go/codec v1.3.1 // indirect
|
||||
github.com/yusufpapurcu/wmi v1.2.4 // indirect
|
||||
go.etcd.io/etcd/api/v3 v3.6.5 // indirect
|
||||
go.uber.org/mock v0.6.0 // indirect
|
||||
go.uber.org/multierr v1.11.0 // indirect
|
||||
@@ -68,6 +71,7 @@ require (
|
||||
github.com/jinzhu/now v1.1.5 // indirect
|
||||
github.com/oklog/ulid/v2 v2.1.1 // indirect
|
||||
github.com/redis/go-redis/v9 v9.16.0 // indirect
|
||||
github.com/shirou/gopsutil/v3 v3.24.5
|
||||
go.etcd.io/etcd/client/pkg/v3 v3.6.5 // indirect
|
||||
go.etcd.io/etcd/client/v3 v3.6.5
|
||||
golang.org/x/crypto v0.43.0 // indirect
|
||||
|
||||
10
go.sum
10
go.sum
@@ -41,6 +41,8 @@ github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI=
|
||||
github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
|
||||
github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
|
||||
github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
|
||||
github.com/go-ole/go-ole v1.2.6 h1:/Fpf6oFPoeFik9ty7siob0G6Ke8QvQEuVcuChpwXzpY=
|
||||
github.com/go-ole/go-ole v1.2.6/go.mod h1:pprOEPIfldk/42T2oK7lQ4v4JSDwmV0As9GaiUsvbm0=
|
||||
github.com/go-playground/assert/v2 v2.2.0 h1:JvknZsQTYeFEAhQwI4qEt9cyV5ONwRHC+lYKSsYSR8s=
|
||||
github.com/go-playground/assert/v2 v2.2.0/go.mod h1:VDjEfimB/XKnb+ZQfWdccd7VUvScMdVu0Titje2rxJ4=
|
||||
github.com/go-playground/locales v0.14.1 h1:EWaQ/wswjilfKLTECiXz7Rh+3BjFhfDFKv/oXslEjJA=
|
||||
@@ -106,6 +108,8 @@ github.com/pelletier/go-toml/v2 v2.2.4 h1:mye9XuhQ6gvn5h28+VilKrrPoQVanw5PMw/TB0
|
||||
github.com/pelletier/go-toml/v2 v2.2.4/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY=
|
||||
github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
|
||||
github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
|
||||
github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c h1:ncq/mPwQF4JjgDlrVEn3C11VoGHZN7m8qihwgMEtzYw=
|
||||
github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c/go.mod h1:OmDBASR4679mdNQnz2pUhc2G8CO2JrUAVFDRBDP/hJE=
|
||||
github.com/quic-go/qpack v0.5.1 h1:giqksBPnT/HDtZ6VhtFKgoLOWmlyo9Ei6u9PqzIMbhI=
|
||||
github.com/quic-go/qpack v0.5.1/go.mod h1:+PC4XFrEskIVkcLzpEkbLqq1uCoxPhQuvK5rH1ZgaEg=
|
||||
github.com/quic-go/quic-go v0.55.0 h1:zccPQIqYCXDt5NmcEabyYvOnomjs8Tlwl7tISjJh9Mk=
|
||||
@@ -114,6 +118,8 @@ github.com/redis/go-redis/v9 v9.16.0 h1:OotgqgLSRCmzfqChbQyG1PHC3tLNR89DG4jdOERS
|
||||
github.com/redis/go-redis/v9 v9.16.0/go.mod h1:u410H11HMLoB+TP67dz8rL9s6QW2j76l0//kSOd3370=
|
||||
github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ=
|
||||
github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc=
|
||||
github.com/shirou/gopsutil/v3 v3.24.5 h1:i0t8kL+kQTvpAYToeuiVk3TgDeKOFioZO3Ztz/iZ9pI=
|
||||
github.com/shirou/gopsutil/v3 v3.24.5/go.mod h1:bsoOS1aStSs9ErQ1WWfxllSeS1K5D+U30r2NfcubMVk=
|
||||
github.com/skip2/go-qrcode v0.0.0-20200617195104-da1b6568686e h1:MRM5ITcdelLK2j1vwZ3Je0FKVCfqOLp5zO6trqMLYs0=
|
||||
github.com/skip2/go-qrcode v0.0.0-20200617195104-da1b6568686e/go.mod h1:XV66xRDqSt+GTGFMVlhk3ULuV0y9ZmzeVGR4mloJI3M=
|
||||
github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
|
||||
@@ -134,6 +140,8 @@ github.com/ugorji/go/codec v1.3.1 h1:waO7eEiFDwidsBN6agj1vJQ4AG7lh2yqXyOXqhgQuyY
|
||||
github.com/ugorji/go/codec v1.3.1/go.mod h1:pRBVtBSKl77K30Bv8R2P+cLSGaTtex6fsA2Wjqmfxj4=
|
||||
github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
|
||||
github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74=
|
||||
github.com/yusufpapurcu/wmi v1.2.4 h1:zFUKzehAFReQwLys1b/iSMl+JQGSCSjtVqQn9bBrPo0=
|
||||
github.com/yusufpapurcu/wmi v1.2.4/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0=
|
||||
go.etcd.io/etcd/api/v3 v3.6.5 h1:pMMc42276sgR1j1raO/Qv3QI9Af/AuyQUW6CBAWuntA=
|
||||
go.etcd.io/etcd/api/v3 v3.6.5/go.mod h1:ob0/oWA/UQQlT1BmaEkWQzI0sJ1M0Et0mMpaABxguOQ=
|
||||
go.etcd.io/etcd/client/pkg/v3 v3.6.5 h1:Duz9fAzIZFhYWgRjp/FgNq2gO1jId9Yae/rLn3RrBP8=
|
||||
@@ -184,7 +192,9 @@ golang.org/x/sync v0.17.0 h1:l60nONMj9l5drqw6jlhIELNv9I0A4OFgRsG9k2oT9Ug=
|
||||
golang.org/x/sync v0.17.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI=
|
||||
golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
|
||||
golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.0.0-20201204225414-ed752295db88/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
|
||||
golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
|
||||
golang.org/x/sys v0.37.0 h1:fdNQudmxPjkdUTPnLn5mdQv7Zwvbvpaxqs831goi9kQ=
|
||||
golang.org/x/sys v0.37.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks=
|
||||
|
||||
@@ -1,8 +1,11 @@
|
||||
package config
|
||||
|
||||
import (
|
||||
"log"
|
||||
"fmt"
|
||||
"math"
|
||||
"net"
|
||||
"net/url"
|
||||
"path/filepath"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"strings"
|
||||
@@ -20,21 +23,32 @@ type SrvConfig struct {
|
||||
Databases *conf.DBConf `yaml:"Databases"`
|
||||
MicroService *conf.MicroServiceConf `yaml:"MicroService"`
|
||||
Etcd *conf.EtcdConf `yaml:"Etcd"`
|
||||
ObjectStorage ObjectStorageConf `yaml:"ObjectStorage"`
|
||||
Storage StorageConf `yaml:"Storage"`
|
||||
Namespaces map[string]NamespaceConf `yaml:"Namespaces"`
|
||||
ServiceClients map[string]string `yaml:"ServiceClients"`
|
||||
Cleanup CleanupConf `yaml:"Cleanup"`
|
||||
}
|
||||
|
||||
type ObjectStorageConf struct {
|
||||
type StorageConf struct {
|
||||
Provider string `yaml:"Provider"`
|
||||
ExternalBaseURL string `yaml:"ExternalBaseURL"`
|
||||
AccessTTLSeconds int64 `yaml:"AccessTTLSeconds"`
|
||||
SigningSecret string `yaml:"SigningSecret"`
|
||||
Local LocalStorageConf `yaml:"Local"`
|
||||
Aliyun AliyunStorageConf `yaml:"Aliyun"`
|
||||
}
|
||||
|
||||
type LocalStorageConf struct {
|
||||
RootPath string `yaml:"RootPath"`
|
||||
MinimumFreeSpaceMB uint64 `yaml:"MinimumFreeSpaceMB"`
|
||||
}
|
||||
|
||||
type AliyunStorageConf struct {
|
||||
Endpoint string `yaml:"Endpoint"`
|
||||
Region string `yaml:"Region"`
|
||||
Bucket string `yaml:"Bucket"`
|
||||
PublicBaseURL string `yaml:"PublicBaseURL"`
|
||||
AccessKeyID string `yaml:"AccessKeyID"`
|
||||
AccessKeySecret string `yaml:"AccessKeySecret"`
|
||||
PresignTTLSeconds int64 `yaml:"PresignTTLSeconds"`
|
||||
}
|
||||
|
||||
type NamespaceConf struct {
|
||||
@@ -45,59 +59,110 @@ type NamespaceConf struct {
|
||||
|
||||
type CleanupConf struct {
|
||||
IntervalSeconds int64 `yaml:"IntervalSeconds"`
|
||||
IntegrityIntervalSeconds int64 `yaml:"IntegrityIntervalSeconds"`
|
||||
IntegrityBatchSize int `yaml:"IntegrityBatchSize"`
|
||||
}
|
||||
|
||||
func New(srvKey string) {
|
||||
conf.New(srvKey, &Spec)
|
||||
|
||||
Spec.Port = conf.CheckPort(Spec.Port)
|
||||
Spec.BindIP = conf.CheckIP(Spec.BindIP)
|
||||
if err := Validate(); err != nil {
|
||||
panic(fmt.Errorf("files 配置校验失败: %w", err))
|
||||
}
|
||||
Spec.Addr = net.JoinHostPort(Spec.BindIP, Spec.Port)
|
||||
|
||||
validate()
|
||||
conf.PrintInfo(Spec.Addr)
|
||||
}
|
||||
|
||||
func validate() {
|
||||
validateRequired("Service", Spec.Service)
|
||||
validateRequired("Cache", Spec.Cache)
|
||||
|
||||
if Spec.ObjectStorage.Provider != "aliyun" {
|
||||
configError("ObjectStorage.Provider", "必须为 aliyun")
|
||||
// Validate 校验文件服务配置,不生成随机监听参数或对象存储默认值。
|
||||
func Validate() error {
|
||||
Spec.Service = strings.TrimSpace(Spec.Service)
|
||||
Spec.Port = strings.TrimSpace(Spec.Port)
|
||||
Spec.BindIP = strings.TrimSpace(Spec.BindIP)
|
||||
if err := validateRequired("Service", Spec.Service); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := validateRequired("Cache", Spec.Cache); err != nil {
|
||||
return err
|
||||
}
|
||||
port, err := strconv.Atoi(Spec.Port)
|
||||
if err != nil || port < 1 || port > 65535 {
|
||||
return fmt.Errorf("配置项 Port 无效:必须是 1 到 65535 的整数")
|
||||
}
|
||||
if net.ParseIP(Spec.BindIP) == nil {
|
||||
return fmt.Errorf("配置项 BindIP 无效:必须是明确的 IPv4 或 IPv6 地址")
|
||||
}
|
||||
if Spec.Databases == nil || strings.TrimSpace(Spec.Databases.Driver) == "" || len(Spec.Databases.Source) == 0 {
|
||||
return fmt.Errorf("配置项 Databases 无效:Driver 和 Source 不能为空")
|
||||
}
|
||||
|
||||
validateRequired("ObjectStorage.Endpoint", Spec.ObjectStorage.Endpoint)
|
||||
validateRequired("ObjectStorage.Region", Spec.ObjectStorage.Region)
|
||||
validateRequired("ObjectStorage.Bucket", Spec.ObjectStorage.Bucket)
|
||||
validateRequired("ObjectStorage.PublicBaseURL", Spec.ObjectStorage.PublicBaseURL)
|
||||
validateRequired("ObjectStorage.AccessKeyID", Spec.ObjectStorage.AccessKeyID)
|
||||
validateRequired("ObjectStorage.AccessKeySecret", Spec.ObjectStorage.AccessKeySecret)
|
||||
Spec.Storage.Provider = strings.ToLower(strings.TrimSpace(Spec.Storage.Provider))
|
||||
for path, value := range map[string]string{
|
||||
"Storage.Provider": Spec.Storage.Provider,
|
||||
"Storage.ExternalBaseURL": Spec.Storage.ExternalBaseURL,
|
||||
"Storage.SigningSecret": Spec.Storage.SigningSecret,
|
||||
} {
|
||||
if err := validateRequired(path, value); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if err := validateHTTPURL("Storage.ExternalBaseURL", Spec.Storage.ExternalBaseURL); err != nil {
|
||||
return err
|
||||
}
|
||||
if Spec.Storage.AccessTTLSeconds < 60 || Spec.Storage.AccessTTLSeconds > 3600 {
|
||||
return configError("Storage.AccessTTLSeconds", "必须在 60 到 3600 秒之间")
|
||||
}
|
||||
|
||||
if Spec.ObjectStorage.PresignTTLSeconds < 60 || Spec.ObjectStorage.PresignTTLSeconds > 3600 {
|
||||
configError("ObjectStorage.PresignTTLSeconds", "必须在 60 到 3600 秒之间")
|
||||
switch Spec.Storage.Provider {
|
||||
case "local":
|
||||
Spec.Storage.Local.RootPath = strings.TrimSpace(Spec.Storage.Local.RootPath)
|
||||
if !filepath.IsAbs(Spec.Storage.Local.RootPath) {
|
||||
return configError("Storage.Local.RootPath", "必须是绝对路径")
|
||||
}
|
||||
if Spec.Storage.Local.MinimumFreeSpaceMB == 0 {
|
||||
return configError("Storage.Local.MinimumFreeSpaceMB", "必须大于 0")
|
||||
}
|
||||
if Spec.Storage.Local.MinimumFreeSpaceMB > math.MaxUint64/(1024*1024) {
|
||||
return configError("Storage.Local.MinimumFreeSpaceMB", "换算为字节后溢出")
|
||||
}
|
||||
case "aliyun":
|
||||
for path, value := range map[string]string{
|
||||
"Storage.Aliyun.Endpoint": Spec.Storage.Aliyun.Endpoint,
|
||||
"Storage.Aliyun.Region": Spec.Storage.Aliyun.Region,
|
||||
"Storage.Aliyun.Bucket": Spec.Storage.Aliyun.Bucket,
|
||||
"Storage.Aliyun.AccessKeyID": Spec.Storage.Aliyun.AccessKeyID,
|
||||
"Storage.Aliyun.AccessKeySecret": Spec.Storage.Aliyun.AccessKeySecret,
|
||||
} {
|
||||
if err := validateRequired(path, value); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
if err := validateHTTPURL("Storage.Aliyun.Endpoint", Spec.Storage.Aliyun.Endpoint); err != nil {
|
||||
return err
|
||||
}
|
||||
default:
|
||||
return configError("Storage.Provider", "必须为 local 或 aliyun")
|
||||
}
|
||||
|
||||
if len(Spec.Namespaces) == 0 {
|
||||
configError("Namespaces", "不能为空")
|
||||
return configError("Namespaces", "不能为空")
|
||||
}
|
||||
for name, namespace := range Spec.Namespaces {
|
||||
path := "Namespaces." + name
|
||||
if !configNameRegexp.MatchString(name) {
|
||||
configError(path, "名称仅允许小写字母、数字和连字符")
|
||||
return configError(path, "名称仅允许小写字母、数字和连字符")
|
||||
}
|
||||
if !isSafeRelativeObjectPath(namespace.Prefix) {
|
||||
configError(path+".Prefix", "必须是安全的相对对象路径")
|
||||
return configError(path+".Prefix", "必须是安全的相对对象路径")
|
||||
}
|
||||
if namespace.MaxSizeMB <= 0 {
|
||||
configError(path+".MaxSizeMB", "必须大于 0")
|
||||
return configError(path+".MaxSizeMB", "必须大于 0")
|
||||
}
|
||||
if len(namespace.AllowedExtensions) == 0 {
|
||||
configError(path+".AllowedExtensions", "不能为空")
|
||||
return configError(path+".AllowedExtensions", "不能为空")
|
||||
}
|
||||
for index, extension := range namespace.AllowedExtensions {
|
||||
extensionPath := path + ".AllowedExtensions[" + stringIndex(index) + "]"
|
||||
if !strings.HasPrefix(extension, ".") {
|
||||
configError(extensionPath, "必须以 . 开头")
|
||||
return configError(extensionPath, "必须以 . 开头")
|
||||
}
|
||||
namespace.AllowedExtensions[index] = strings.ToLower(extension)
|
||||
}
|
||||
@@ -105,25 +170,43 @@ func validate() {
|
||||
}
|
||||
|
||||
if len(Spec.ServiceClients) == 0 {
|
||||
configError("ServiceClients", "不能为空")
|
||||
return configError("ServiceClients", "不能为空")
|
||||
}
|
||||
for name, secret := range Spec.ServiceClients {
|
||||
path := "ServiceClients." + name
|
||||
if !configNameRegexp.MatchString(name) {
|
||||
configError(path, "服务名仅允许小写字母、数字和连字符")
|
||||
return configError(path, "服务名仅允许小写字母、数字和连字符")
|
||||
}
|
||||
if err := validateRequired(path, secret); err != nil {
|
||||
return err
|
||||
}
|
||||
validateRequired(path, secret)
|
||||
}
|
||||
|
||||
if Spec.Cleanup.IntervalSeconds <= 0 {
|
||||
configError("Cleanup.IntervalSeconds", "必须大于 0")
|
||||
return configError("Cleanup.IntervalSeconds", "必须大于 0")
|
||||
}
|
||||
if Spec.Cleanup.IntegrityIntervalSeconds <= 0 {
|
||||
return configError("Cleanup.IntegrityIntervalSeconds", "必须大于 0")
|
||||
}
|
||||
if Spec.Cleanup.IntegrityBatchSize <= 0 || Spec.Cleanup.IntegrityBatchSize > 1000 {
|
||||
return configError("Cleanup.IntegrityBatchSize", "必须在 1 到 1000 之间")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func validateRequired(path, value string) {
|
||||
if strings.TrimSpace(value) == "" {
|
||||
configError(path, "不能为空")
|
||||
func validateHTTPURL(path, raw string) error {
|
||||
parsed, err := url.Parse(strings.TrimSpace(raw))
|
||||
if err != nil || parsed.Host == "" || (parsed.Scheme != "http" && parsed.Scheme != "https") {
|
||||
return configError(path, "必须是完整的 HTTP 或 HTTPS 地址")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func validateRequired(path, value string) error {
|
||||
if strings.TrimSpace(value) == "" {
|
||||
return configError(path, "不能为空")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func isSafeRelativeObjectPath(path string) bool {
|
||||
@@ -138,8 +221,8 @@ func isSafeRelativeObjectPath(path string) bool {
|
||||
return true
|
||||
}
|
||||
|
||||
func configError(path, message string) {
|
||||
log.Fatalf("配置项 %s 无效:%s", path, message)
|
||||
func configError(path, message string) error {
|
||||
return fmt.Errorf("配置项 %s 无效:%s", path, message)
|
||||
}
|
||||
|
||||
func stringIndex(index int) string {
|
||||
|
||||
@@ -10,8 +10,13 @@ var (
|
||||
ErrFileNotFound = errcode.NewError(2105, "文件不存在")
|
||||
ErrUploadStatusConflict = errcode.NewError(2106, "上传状态冲突")
|
||||
ErrUploadExpired = errcode.NewError(2107, "上传已过期")
|
||||
ErrObjectInfoMismatch = errcode.NewError(2108, "OSS 对象信息不匹配")
|
||||
ErrObjectInfoMismatch = errcode.NewError(2108, "文件对象信息不匹配")
|
||||
ErrUnauthorizedOperation = errcode.NewError(2109, "无权操作")
|
||||
ErrObjectStorageOperation = errcode.NewError(2110, "OSS 操作失败")
|
||||
ErrStorageOperation = errcode.NewError(2110, "文件存储操作失败")
|
||||
ErrDatabaseOperation = errcode.NewError(2111, "数据库操作失败")
|
||||
ErrSignatureInvalid = errcode.NewError(2112, "文件访问签名无效或已过期")
|
||||
ErrStorageSpaceLow = errcode.NewError(2113, "文件存储空间不足")
|
||||
ErrVisibilityNotAllowed = errcode.NewError(2114, "该文件不允许公开访问")
|
||||
ErrRangeNotSupported = errcode.NewError(2115, "不支持多段文件读取")
|
||||
ErrUploadLocked = errcode.NewError(2116, "文件正在上传")
|
||||
)
|
||||
|
||||
143
internal/health/readiness.go
Normal file
143
internal/health/readiness.go
Normal file
@@ -0,0 +1,143 @@
|
||||
package health
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"runtime/debug"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.apinb.com/ops/files/internal/config"
|
||||
"git.apinb.com/ops/files/internal/impl"
|
||||
"git.apinb.com/ops/files/internal/jobs"
|
||||
"git.apinb.com/ops/files/internal/runtimeinfo"
|
||||
"git.apinb.com/ops/files/internal/storage"
|
||||
"github.com/gin-gonic/gin"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
type CheckResult struct {
|
||||
Name string `json:"name"`
|
||||
Required bool `json:"required"`
|
||||
Ready bool `json:"ready"`
|
||||
Message string `json:"message,omitempty"`
|
||||
}
|
||||
type Readiness struct {
|
||||
Ready bool `json:"ready"`
|
||||
CheckedAt time.Time `json:"checked_at"`
|
||||
Checks []CheckResult `json:"checks,omitempty"`
|
||||
Version string `json:"version"`
|
||||
}
|
||||
|
||||
type FileCounts struct {
|
||||
Total int64 `json:"total"`
|
||||
Ready int64 `json:"ready"`
|
||||
Pending int64 `json:"pending"`
|
||||
}
|
||||
|
||||
func Evaluate(ctx context.Context) Readiness {
|
||||
result, _ := evaluate(ctx)
|
||||
return result
|
||||
}
|
||||
|
||||
func evaluate(ctx context.Context) (Readiness, storage.RuntimeStatus) {
|
||||
result := Readiness{Ready: true, CheckedAt: time.Now().UTC(), Version: runtimeVersion()}
|
||||
result.add("config", true, config.Validate())
|
||||
result.add("database", true, pingDatabase(ctx, impl.DBService))
|
||||
storageStatus, storageErr := impl.StorageService.Status(ctx)
|
||||
if storageErr == nil && !storageStatus.Writable {
|
||||
storageErr = fmt.Errorf("文件存储当前不可写")
|
||||
}
|
||||
result.add("storage", true, storageErr)
|
||||
worker := jobs.Status()
|
||||
if worker.Running {
|
||||
result.add("pending_upload_cleanup", true, nil)
|
||||
} else {
|
||||
result.add("pending_upload_cleanup", true, fmt.Errorf("过期上传清理任务未运行"))
|
||||
}
|
||||
integrity := jobs.IntegrityStatusSnapshot()
|
||||
if integrity.Running {
|
||||
result.add("integrity_worker", true, nil)
|
||||
} else {
|
||||
result.add("integrity_worker", true, fmt.Errorf("文件完整性任务未运行"))
|
||||
}
|
||||
if integrity.Missing+integrity.SizeMismatch+integrity.ETagMismatch > 0 {
|
||||
result.add("file_integrity", false, fmt.Errorf("发现文件缺失或校验不一致"))
|
||||
} else {
|
||||
result.add("file_integrity", false, nil)
|
||||
}
|
||||
return result, storageStatus
|
||||
}
|
||||
|
||||
func runtimeVersion() string {
|
||||
info, ok := debug.ReadBuildInfo()
|
||||
if !ok || strings.TrimSpace(info.Main.Version) == "" {
|
||||
return "unknown"
|
||||
}
|
||||
return info.Main.Version
|
||||
}
|
||||
func Ready(c *gin.Context) {
|
||||
result := Evaluate(c.Request.Context())
|
||||
status := http.StatusOK
|
||||
if !result.Ready {
|
||||
status = http.StatusServiceUnavailable
|
||||
}
|
||||
c.JSON(status, gin.H{"ready": result.Ready})
|
||||
}
|
||||
func Status(c *gin.Context) {
|
||||
result, storageStatus := evaluate(c.Request.Context())
|
||||
status := http.StatusOK
|
||||
if !result.Ready {
|
||||
status = http.StatusServiceUnavailable
|
||||
}
|
||||
counts, countErr := fileCounts(c.Request.Context())
|
||||
if countErr != nil {
|
||||
result.add("file_ledger", true, countErr)
|
||||
status = http.StatusServiceUnavailable
|
||||
}
|
||||
c.JSON(status, gin.H{
|
||||
"readiness": result,
|
||||
"workers": gin.H{
|
||||
"pending_upload_cleanup": jobs.Status(),
|
||||
"integrity": jobs.IntegrityStatusSnapshot(),
|
||||
},
|
||||
"storage": storageStatus,
|
||||
"disk": storageStatus.Disk,
|
||||
"files": counts,
|
||||
"activity": runtimeinfo.Snapshot(),
|
||||
})
|
||||
}
|
||||
|
||||
func fileCounts(ctx context.Context) (FileCounts, error) {
|
||||
var counts FileCounts
|
||||
err := impl.DBService.WithContext(ctx).Raw(`
|
||||
SELECT COUNT(*) AS total,
|
||||
COUNT(*) FILTER (WHERE status = 'ready') AS ready,
|
||||
COUNT(*) FILTER (WHERE status = 'pending') AS pending
|
||||
FROM files_object
|
||||
WHERE deleted_at IS NULL`).Scan(&counts).Error
|
||||
return counts, err
|
||||
}
|
||||
func (r *Readiness) add(name string, required bool, err error) {
|
||||
check := CheckResult{Name: name, Required: required, Ready: err == nil}
|
||||
if err != nil {
|
||||
check.Message = strings.TrimSpace(err.Error())
|
||||
if required {
|
||||
r.Ready = false
|
||||
}
|
||||
}
|
||||
r.Checks = append(r.Checks, check)
|
||||
}
|
||||
func pingDatabase(ctx context.Context, db *gorm.DB) error {
|
||||
if db == nil {
|
||||
return fmt.Errorf("数据库未初始化")
|
||||
}
|
||||
sqlDB, err := db.DB()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
checkCtx, cancel := context.WithTimeout(ctx, 2*time.Second)
|
||||
defer cancel()
|
||||
return sqlDB.PingContext(checkCtx)
|
||||
}
|
||||
@@ -1,6 +1,8 @@
|
||||
package impl
|
||||
|
||||
import (
|
||||
"errors"
|
||||
|
||||
"git.apinb.com/bsm-sdk/core/cache/redis"
|
||||
"git.apinb.com/bsm-sdk/core/logger"
|
||||
"git.apinb.com/bsm-sdk/core/types"
|
||||
@@ -8,6 +10,7 @@ import (
|
||||
"git.apinb.com/bsm-sdk/core/with"
|
||||
"git.apinb.com/ops/files/internal/config"
|
||||
"git.apinb.com/ops/files/internal/storage"
|
||||
"git.apinb.com/ops/files/internal/storage/factory"
|
||||
clientv3 "go.etcd.io/etcd/client/v3"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
@@ -16,7 +19,7 @@ var (
|
||||
RedisService *redis.RedisClient
|
||||
EtcdService *clientv3.Client
|
||||
DBService *gorm.DB
|
||||
StorageService *storage.Client
|
||||
StorageService storage.Backend
|
||||
Logger *logger.Logger
|
||||
)
|
||||
|
||||
@@ -31,7 +34,7 @@ func NewImpl() {
|
||||
})
|
||||
|
||||
var err error
|
||||
StorageService, err = storage.New(config.Spec.ObjectStorage)
|
||||
StorageService, err = factory.New(config.Spec.Storage)
|
||||
if err != nil {
|
||||
panic(err)
|
||||
}
|
||||
@@ -40,3 +43,26 @@ func NewImpl() {
|
||||
EtcdService = with.Etcd(config.Spec.Etcd)
|
||||
logger.New(nil)
|
||||
}
|
||||
|
||||
// Close 关闭数据库、Redis 和 Etcd 连接。
|
||||
func Close() error {
|
||||
var closeErrors []error
|
||||
if DBService != nil {
|
||||
if sqlDB, err := DBService.DB(); err != nil {
|
||||
closeErrors = append(closeErrors, err)
|
||||
} else if err := sqlDB.Close(); err != nil {
|
||||
closeErrors = append(closeErrors, err)
|
||||
}
|
||||
}
|
||||
if RedisService != nil && RedisService.Client != nil {
|
||||
if err := RedisService.Client.Close(); err != nil {
|
||||
closeErrors = append(closeErrors, err)
|
||||
}
|
||||
}
|
||||
if EtcdService != nil {
|
||||
if err := EtcdService.Close(); err != nil {
|
||||
closeErrors = append(closeErrors, err)
|
||||
}
|
||||
}
|
||||
return errors.Join(closeErrors...)
|
||||
}
|
||||
|
||||
151
internal/jobs/integrity.go
Normal file
151
internal/jobs/integrity.go
Normal file
@@ -0,0 +1,151 @@
|
||||
package jobs
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.apinb.com/bsm-sdk/core/logger"
|
||||
"git.apinb.com/ops/files/internal/config"
|
||||
"git.apinb.com/ops/files/internal/impl"
|
||||
"git.apinb.com/ops/files/internal/models"
|
||||
"git.apinb.com/ops/files/internal/storage"
|
||||
)
|
||||
|
||||
type IntegrityStatus struct {
|
||||
Running bool `json:"running"`
|
||||
LastStarted time.Time `json:"last_started,omitempty"`
|
||||
LastSuccess time.Time `json:"last_success,omitempty"`
|
||||
LastError string `json:"last_error,omitempty"`
|
||||
Checked int64 `json:"checked"`
|
||||
Missing int64 `json:"missing"`
|
||||
SizeMismatch int64 `json:"size_mismatch"`
|
||||
ETagMismatch int64 `json:"etag_mismatch"`
|
||||
}
|
||||
|
||||
var integrityRuntime struct {
|
||||
sync.RWMutex
|
||||
IntegrityStatus
|
||||
wg sync.WaitGroup
|
||||
}
|
||||
|
||||
func StartIntegrity(ctx context.Context) error {
|
||||
integrityRuntime.Lock()
|
||||
if integrityRuntime.Running {
|
||||
integrityRuntime.Unlock()
|
||||
return fmt.Errorf("文件完整性任务已启动")
|
||||
}
|
||||
integrityRuntime.Unlock()
|
||||
if err := runIntegrity(ctx); err != nil {
|
||||
return fmt.Errorf("首次核对文件完整性失败: %w", err)
|
||||
}
|
||||
integrityRuntime.Lock()
|
||||
integrityRuntime.Running = true
|
||||
integrityRuntime.Unlock()
|
||||
integrityRuntime.wg.Add(1)
|
||||
go func() {
|
||||
defer integrityRuntime.wg.Done()
|
||||
defer func() {
|
||||
integrityRuntime.Lock()
|
||||
integrityRuntime.Running = false
|
||||
integrityRuntime.Unlock()
|
||||
}()
|
||||
ticker := time.NewTicker(time.Duration(config.Spec.Cleanup.IntegrityIntervalSeconds) * time.Second)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
if err := runIntegrity(ctx); err != nil && !errors.Is(err, context.Canceled) {
|
||||
logger.Errorf("stage=files_integrity error=%v", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
return nil
|
||||
}
|
||||
|
||||
func IntegrityStatusSnapshot() IntegrityStatus {
|
||||
integrityRuntime.RLock()
|
||||
defer integrityRuntime.RUnlock()
|
||||
return integrityRuntime.IntegrityStatus
|
||||
}
|
||||
|
||||
func waitIntegrity(ctx context.Context) error {
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
integrityRuntime.wg.Wait()
|
||||
close(done)
|
||||
}()
|
||||
select {
|
||||
case <-done:
|
||||
return nil
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
func runIntegrity(ctx context.Context) error {
|
||||
startedAt := time.Now().UTC()
|
||||
status := IntegrityStatus{Running: true, LastStarted: startedAt}
|
||||
integrityRuntime.Lock()
|
||||
status.LastSuccess = integrityRuntime.LastSuccess
|
||||
integrityRuntime.IntegrityStatus = status
|
||||
integrityRuntime.Unlock()
|
||||
|
||||
var lastID uint
|
||||
for {
|
||||
var fileObjects []models.FileObject
|
||||
result := impl.DBService.WithContext(ctx).
|
||||
Where("status = ? AND id > ?", models.FileStatusReady, lastID).
|
||||
Order("id ASC").Limit(config.Spec.Cleanup.IntegrityBatchSize).Find(&fileObjects)
|
||||
if result.Error != nil {
|
||||
return finishIntegrity(status, result.Error)
|
||||
}
|
||||
if len(fileObjects) == 0 {
|
||||
break
|
||||
}
|
||||
for _, fileObject := range fileObjects {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return finishIntegrity(status, err)
|
||||
}
|
||||
status.Checked++
|
||||
objectInfo, err := impl.StorageService.Stat(ctx, fileObject.ObjectKey)
|
||||
if err != nil {
|
||||
if errors.Is(err, storage.ErrObjectNotFound) {
|
||||
status.Missing++
|
||||
continue
|
||||
}
|
||||
return finishIntegrity(status, err)
|
||||
}
|
||||
if objectInfo.Size != fileObject.ActualSize {
|
||||
status.SizeMismatch++
|
||||
}
|
||||
if objectInfo.StorageETag != fileObject.StorageETag {
|
||||
status.ETagMismatch++
|
||||
}
|
||||
}
|
||||
lastID = fileObjects[len(fileObjects)-1].ID
|
||||
if len(fileObjects) < config.Spec.Cleanup.IntegrityBatchSize {
|
||||
break
|
||||
}
|
||||
}
|
||||
status.LastSuccess = time.Now().UTC()
|
||||
status.Running = integrityRuntime.Running
|
||||
integrityRuntime.Lock()
|
||||
integrityRuntime.IntegrityStatus = status
|
||||
integrityRuntime.Unlock()
|
||||
return nil
|
||||
}
|
||||
|
||||
func finishIntegrity(status IntegrityStatus, err error) error {
|
||||
status.LastError = err.Error()
|
||||
status.Running = integrityRuntime.Running
|
||||
integrityRuntime.Lock()
|
||||
integrityRuntime.IntegrityStatus = status
|
||||
integrityRuntime.Unlock()
|
||||
return err
|
||||
}
|
||||
@@ -2,6 +2,9 @@ package jobs
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"git.apinb.com/bsm-sdk/core/logger"
|
||||
@@ -9,24 +12,108 @@ import (
|
||||
"git.apinb.com/ops/files/internal/impl"
|
||||
"git.apinb.com/ops/files/internal/lifecycle"
|
||||
"git.apinb.com/ops/files/internal/models"
|
||||
"git.apinb.com/ops/files/internal/storage"
|
||||
)
|
||||
|
||||
func StartPendingUploadCleanup() {
|
||||
// CleanupStatus 描述过期上传清理任务的运行状态。
|
||||
type CleanupStatus struct {
|
||||
Running bool `json:"running"`
|
||||
LastStarted time.Time `json:"last_started,omitempty"`
|
||||
LastSuccess time.Time `json:"last_success,omitempty"`
|
||||
LastError string `json:"last_error,omitempty"`
|
||||
}
|
||||
|
||||
var cleanupRuntime struct {
|
||||
sync.RWMutex
|
||||
CleanupStatus
|
||||
wg sync.WaitGroup
|
||||
}
|
||||
|
||||
// StartPendingUploadCleanup 启动可取消的过期上传清理任务。
|
||||
func StartPendingUploadCleanup(ctx context.Context) error {
|
||||
cleanupRuntime.Lock()
|
||||
if cleanupRuntime.Running {
|
||||
cleanupRuntime.Unlock()
|
||||
return fmt.Errorf("过期上传清理任务已启动")
|
||||
}
|
||||
cleanupRuntime.Unlock()
|
||||
|
||||
if err := runCleanup(ctx); err != nil {
|
||||
return fmt.Errorf("首次清理过期上传失败: %w", err)
|
||||
}
|
||||
|
||||
cleanupRuntime.Lock()
|
||||
cleanupRuntime.Running = true
|
||||
cleanupRuntime.Unlock()
|
||||
cleanupRuntime.wg.Add(1)
|
||||
go func() {
|
||||
cleanupPendingUploads()
|
||||
defer cleanupRuntime.wg.Done()
|
||||
defer func() {
|
||||
cleanupRuntime.Lock()
|
||||
cleanupRuntime.Running = false
|
||||
cleanupRuntime.Unlock()
|
||||
}()
|
||||
|
||||
ticker := time.NewTicker(time.Duration(config.Spec.Cleanup.IntervalSeconds) * time.Second)
|
||||
defer ticker.Stop()
|
||||
for range ticker.C {
|
||||
cleanupPendingUploads()
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-ticker.C:
|
||||
if err := runCleanup(ctx); err != nil && !errors.Is(err, context.Canceled) {
|
||||
logger.Errorf("stage=cleanup error=%v", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
return nil
|
||||
}
|
||||
|
||||
func cleanupPendingUploads() {
|
||||
// Status 返回清理任务状态快照。
|
||||
func Status() CleanupStatus {
|
||||
cleanupRuntime.RLock()
|
||||
defer cleanupRuntime.RUnlock()
|
||||
return cleanupRuntime.CleanupStatus
|
||||
}
|
||||
|
||||
// Wait 等待清理任务退出。
|
||||
func Wait(ctx context.Context) error {
|
||||
done := make(chan struct{})
|
||||
go func() {
|
||||
cleanupRuntime.wg.Wait()
|
||||
close(done)
|
||||
}()
|
||||
select {
|
||||
case <-done:
|
||||
return waitIntegrity(ctx)
|
||||
case <-ctx.Done():
|
||||
return ctx.Err()
|
||||
}
|
||||
}
|
||||
|
||||
func runCleanup(ctx context.Context) error {
|
||||
startedAt := time.Now().UTC()
|
||||
cleanupRuntime.Lock()
|
||||
cleanupRuntime.LastStarted = startedAt
|
||||
cleanupRuntime.Unlock()
|
||||
|
||||
err := cleanupPendingUploads(ctx)
|
||||
cleanupRuntime.Lock()
|
||||
defer cleanupRuntime.Unlock()
|
||||
if err != nil {
|
||||
cleanupRuntime.LastError = err.Error()
|
||||
return err
|
||||
}
|
||||
cleanupRuntime.LastSuccess = time.Now().UTC()
|
||||
cleanupRuntime.LastError = ""
|
||||
return nil
|
||||
}
|
||||
|
||||
func cleanupPendingUploads(ctx context.Context) error {
|
||||
var fileObjects []models.FileObject
|
||||
now := time.Now()
|
||||
result := impl.DBService.
|
||||
result := impl.DBService.WithContext(ctx).
|
||||
Where(
|
||||
"(status = ? AND expires_at <= ?) OR (status = ? AND (delete_lease_until IS NULL OR delete_lease_until <= ?))",
|
||||
models.FileStatusPending,
|
||||
@@ -38,10 +125,10 @@ func cleanupPendingUploads() {
|
||||
Limit(100).
|
||||
Find(&fileObjects)
|
||||
if result.Error != nil {
|
||||
logger.Error("stage=scan")
|
||||
return
|
||||
return fmt.Errorf("扫描待清理文件失败: %w", result.Error)
|
||||
}
|
||||
|
||||
var cleanupErrors []error
|
||||
for _, fileObject := range fileObjects {
|
||||
var lease lifecycle.DeletionLease
|
||||
var claimed bool
|
||||
@@ -54,26 +141,37 @@ func cleanupPendingUploads() {
|
||||
}
|
||||
if err != nil {
|
||||
logger.Errorf("identity=%s stage=claim", fileObject.Identity)
|
||||
cleanupErrors = append(cleanupErrors, fmt.Errorf("文件 %s 获取删除租约失败: %w", fileObject.Identity, err))
|
||||
continue
|
||||
}
|
||||
if !claimed {
|
||||
continue
|
||||
}
|
||||
|
||||
cleanupClaimedFile(fileObject, lease)
|
||||
if err := cleanupClaimedFile(ctx, fileObject, lease); err != nil {
|
||||
cleanupErrors = append(cleanupErrors, err)
|
||||
}
|
||||
}
|
||||
return errors.Join(cleanupErrors...)
|
||||
}
|
||||
|
||||
func cleanupClaimedFile(fileObject models.FileObject, lease lifecycle.DeletionLease) {
|
||||
operationCtx, cancel := context.WithTimeout(context.Background(), lifecycle.DeleteOperationTimeout)
|
||||
func cleanupClaimedFile(ctx context.Context, fileObject models.FileObject, lease lifecycle.DeletionLease) error {
|
||||
operationCtx, cancel := context.WithTimeout(ctx, lifecycle.DeleteOperationTimeout)
|
||||
defer cancel()
|
||||
|
||||
if receiver, ok := impl.StorageService.(storage.UploadReceiver); ok {
|
||||
if err := receiver.DiscardUpload(operationCtx, fileObject.Identity); err != nil {
|
||||
return fmt.Errorf("文件 %s 清理暂存内容失败: %w", fileObject.Identity, err)
|
||||
}
|
||||
}
|
||||
if err := impl.StorageService.Delete(operationCtx, fileObject.ObjectKey); err != nil {
|
||||
logger.Errorf("identity=%s stage=oss_delete", fileObject.Identity)
|
||||
return
|
||||
logger.Errorf("identity=%s stage=storage_delete", fileObject.Identity)
|
||||
return fmt.Errorf("文件 %s 删除对象失败: %w", fileObject.Identity, err)
|
||||
}
|
||||
|
||||
if err := lifecycle.FinalizeDeletion(impl.DBService.WithContext(operationCtx), fileObject.ID, lease.Token); err != nil {
|
||||
logger.Errorf("identity=%s stage=finalize", fileObject.Identity)
|
||||
return fmt.Errorf("文件 %s 完成删除失败: %w", fileObject.Identity, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
154
internal/logic/files/access.go
Normal file
154
internal/logic/files/access.go
Normal file
@@ -0,0 +1,154 @@
|
||||
package files
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.apinb.com/bsm-sdk/core/infra"
|
||||
"git.apinb.com/ops/files/internal/auth"
|
||||
"git.apinb.com/ops/files/internal/config"
|
||||
fileerrors "git.apinb.com/ops/files/internal/errors"
|
||||
"git.apinb.com/ops/files/internal/impl"
|
||||
"git.apinb.com/ops/files/internal/models"
|
||||
"git.apinb.com/ops/files/internal/signing"
|
||||
"github.com/gin-gonic/gin"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
func Access(ctx *gin.Context) {
|
||||
fileObject, ok := loadOwnedFile(ctx)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
if fileObject.Status != models.FileStatusReady {
|
||||
infra.Response.Error(ctx, fileerrors.ErrFileNotFound)
|
||||
return
|
||||
}
|
||||
infra.Response.Success(ctx, newAccessInstruction(fileObject, normalizeDisposition(ctx.Query("disposition"), "inline")))
|
||||
}
|
||||
|
||||
func BatchAccess(ctx *gin.Context) {
|
||||
actor, ok := auth.FromContext(ctx)
|
||||
if !ok || actor.Type != auth.ActorTypeService {
|
||||
infra.Response.Error(ctx, fileerrors.ErrUnauthorizedOperation)
|
||||
return
|
||||
}
|
||||
var request BatchAccessRequest
|
||||
if err := ctx.ShouldBindJSON(&request); err != nil || len(request.FileIDs) == 0 || len(request.FileIDs) > 100 {
|
||||
infra.Response.Error(ctx, fileerrors.ErrInvalidParameter)
|
||||
return
|
||||
}
|
||||
fileIDs := uniqueFileIDs(request.FileIDs)
|
||||
if len(fileIDs) != len(request.FileIDs) {
|
||||
infra.Response.Error(ctx, fileerrors.ErrInvalidParameter)
|
||||
return
|
||||
}
|
||||
var fileObjects []models.FileObject
|
||||
if err := impl.DBService.Where(
|
||||
"identity IN ? AND owner_type = ? AND owner_identity = ? AND status = ?",
|
||||
fileIDs, auth.ActorTypeService, actor.Identity, models.FileStatusReady,
|
||||
).Find(&fileObjects).Error; err != nil {
|
||||
infra.Response.Error(ctx, fileerrors.ErrDatabaseOperation)
|
||||
return
|
||||
}
|
||||
if len(fileObjects) != len(fileIDs) {
|
||||
infra.Response.Error(ctx, fileerrors.ErrFileNotFound)
|
||||
return
|
||||
}
|
||||
byIdentity := make(map[string]models.FileObject, len(fileObjects))
|
||||
for _, fileObject := range fileObjects {
|
||||
byIdentity[fileObject.Identity] = fileObject
|
||||
}
|
||||
disposition := normalizeDisposition(request.Disposition, "inline")
|
||||
response := make([]AccessInstruction, 0, len(fileIDs))
|
||||
for _, fileID := range fileIDs {
|
||||
response = append(response, newAccessInstruction(byIdentity[fileID], disposition))
|
||||
}
|
||||
infra.Response.Success(ctx, response)
|
||||
}
|
||||
|
||||
func SignedAccess(ctx *gin.Context) {
|
||||
fileID := strings.TrimSpace(ctx.Param("identity"))
|
||||
disposition := normalizeDisposition(ctx.Query("disposition"), "inline")
|
||||
expiresUnix, err := strconv.ParseInt(ctx.Query("expires"), 10, 64)
|
||||
if err != nil || !signing.VerifyAccess(
|
||||
config.Spec.Storage.SigningSecret, fileID, time.Unix(expiresUnix, 0), disposition,
|
||||
ctx.Query("signature"), time.Now(),
|
||||
) {
|
||||
ctx.JSON(http.StatusForbidden, gin.H{"message": "文件访问签名无效或已过期"})
|
||||
return
|
||||
}
|
||||
fileObject, ok := loadReadyFile(ctx, fileID, "")
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
streamFile(ctx, fileObject, disposition)
|
||||
}
|
||||
|
||||
func PublicAccess(ctx *gin.Context) {
|
||||
fileObject, ok := loadReadyFile(ctx, strings.TrimSpace(ctx.Param("identity")), models.FileVisibilityPublic)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
streamFile(ctx, fileObject, normalizeDisposition(ctx.Query("disposition"), "inline"))
|
||||
}
|
||||
|
||||
func newAccessInstruction(fileObject models.FileObject, disposition string) AccessInstruction {
|
||||
baseURL := strings.TrimRight(config.Spec.Storage.ExternalBaseURL, "/")
|
||||
result := AccessInstruction{
|
||||
FileID: fileObject.Identity, Filename: fileObject.OriginalName, ContentType: fileObject.ContentType,
|
||||
Size: fileObject.ActualSize, Disposition: disposition,
|
||||
}
|
||||
if fileObject.Visibility == models.FileVisibilityPublic {
|
||||
result.URL = fmt.Sprintf("%s/v1/public/files/%s/content?disposition=%s", baseURL, fileObject.Identity, disposition)
|
||||
return result
|
||||
}
|
||||
expiresAt := time.Now().Add(time.Duration(config.Spec.Storage.AccessTTLSeconds) * time.Second).UTC()
|
||||
signature := signing.SignAccess(config.Spec.Storage.SigningSecret, fileObject.Identity, expiresAt, disposition)
|
||||
result.ExpiresAt = expiresAt
|
||||
result.URL = fmt.Sprintf("%s/v1/access/%s/content?expires=%d&disposition=%s&signature=%s",
|
||||
baseURL, fileObject.Identity, expiresAt.Unix(), disposition, signature)
|
||||
return result
|
||||
}
|
||||
|
||||
func loadReadyFile(ctx *gin.Context, fileID string, visibility models.FileVisibility) (models.FileObject, bool) {
|
||||
if fileID == "" {
|
||||
ctx.Status(http.StatusNotFound)
|
||||
return models.FileObject{}, false
|
||||
}
|
||||
query := impl.DBService.Where("identity = ? AND status = ?", fileID, models.FileStatusReady)
|
||||
if visibility != "" {
|
||||
query = query.Where("visibility = ?", visibility)
|
||||
}
|
||||
var fileObject models.FileObject
|
||||
if err := query.First(&fileObject).Error; err != nil {
|
||||
if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
ctx.Status(http.StatusNotFound)
|
||||
} else {
|
||||
ctx.Status(http.StatusInternalServerError)
|
||||
}
|
||||
return models.FileObject{}, false
|
||||
}
|
||||
return fileObject, true
|
||||
}
|
||||
|
||||
func uniqueFileIDs(values []string) []string {
|
||||
seen := make(map[string]struct{}, len(values))
|
||||
result := make([]string, 0, len(values))
|
||||
for _, value := range values {
|
||||
value = strings.TrimSpace(value)
|
||||
if value == "" {
|
||||
continue
|
||||
}
|
||||
if _, exists := seen[value]; exists {
|
||||
continue
|
||||
}
|
||||
seen[value] = struct{}{}
|
||||
result = append(result, value)
|
||||
}
|
||||
return result
|
||||
}
|
||||
@@ -7,10 +7,19 @@ import (
|
||||
fileerrors "git.apinb.com/ops/files/internal/errors"
|
||||
"git.apinb.com/ops/files/internal/impl"
|
||||
"git.apinb.com/ops/files/internal/models"
|
||||
"git.apinb.com/ops/files/internal/runtimeinfo"
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
func CompleteUpload(ctx *gin.Context) {
|
||||
success := false
|
||||
defer func() {
|
||||
if success {
|
||||
runtimeinfo.RecordUploadSuccess()
|
||||
} else {
|
||||
runtimeinfo.RecordUploadFailure()
|
||||
}
|
||||
}()
|
||||
fileObject, ok := loadOwnedFile(ctx)
|
||||
if !ok {
|
||||
return
|
||||
@@ -27,7 +36,7 @@ func CompleteUpload(ctx *gin.Context) {
|
||||
|
||||
objectInfo, err := impl.StorageService.Stat(ctx.Request.Context(), fileObject.ObjectKey)
|
||||
if err != nil {
|
||||
infra.Response.Error(ctx, fileerrors.ErrObjectStorageOperation)
|
||||
infra.Response.Error(ctx, fileerrors.ErrStorageOperation)
|
||||
return
|
||||
}
|
||||
if objectInfo.Size != fileObject.ExpectedSize {
|
||||
@@ -41,7 +50,7 @@ func CompleteUpload(ctx *gin.Context) {
|
||||
Updates(map[string]any{
|
||||
"status": models.FileStatusReady,
|
||||
"actual_size": objectInfo.Size,
|
||||
"e_tag": objectInfo.ETag,
|
||||
"storage_etag": objectInfo.StorageETag,
|
||||
"completed_at": completedAt,
|
||||
})
|
||||
if result.Error != nil {
|
||||
@@ -55,7 +64,8 @@ func CompleteUpload(ctx *gin.Context) {
|
||||
|
||||
fileObject.Status = models.FileStatusReady
|
||||
fileObject.ActualSize = objectInfo.Size
|
||||
fileObject.ETag = objectInfo.ETag
|
||||
fileObject.StorageETag = objectInfo.StorageETag
|
||||
fileObject.CompletedAt = &completedAt
|
||||
success = true
|
||||
infra.Response.Success(ctx, newFileResponse(fileObject))
|
||||
}
|
||||
|
||||
@@ -9,6 +9,7 @@ import (
|
||||
"git.apinb.com/ops/files/internal/impl"
|
||||
"git.apinb.com/ops/files/internal/lifecycle"
|
||||
"git.apinb.com/ops/files/internal/models"
|
||||
"git.apinb.com/ops/files/internal/runtimeinfo"
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
@@ -37,14 +38,17 @@ func Delete(ctx *gin.Context) {
|
||||
defer cancel()
|
||||
|
||||
if err := impl.StorageService.Delete(operationCtx, fileObject.ObjectKey); err != nil {
|
||||
infra.Response.Error(ctx, fileerrors.ErrObjectStorageOperation)
|
||||
runtimeinfo.RecordDeleteFailure()
|
||||
infra.Response.Error(ctx, fileerrors.ErrStorageOperation)
|
||||
return
|
||||
}
|
||||
|
||||
if err := lifecycle.FinalizeDeletion(impl.DBService.WithContext(operationCtx), fileObject.ID, lease.Token); err != nil {
|
||||
runtimeinfo.RecordDeleteFailure()
|
||||
infra.Response.Error(ctx, fileerrors.ErrDatabaseOperation)
|
||||
return
|
||||
}
|
||||
|
||||
runtimeinfo.RecordDeleteSuccess()
|
||||
infra.Response.Success(ctx, nil)
|
||||
}
|
||||
|
||||
@@ -26,6 +26,19 @@ func Detail(ctx *gin.Context) {
|
||||
infra.Response.Success(ctx, newFileResponse(fileObject))
|
||||
}
|
||||
|
||||
// Download 通过文件服务鉴权读取文件内容。
|
||||
func Download(ctx *gin.Context) {
|
||||
fileObject, ok := loadOwnedFile(ctx)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
if fileObject.Status != models.FileStatusReady {
|
||||
infra.Response.Error(ctx, fileerrors.ErrFileNotFound)
|
||||
return
|
||||
}
|
||||
streamFile(ctx, fileObject, normalizeDisposition(ctx.Query("disposition"), "attachment"))
|
||||
}
|
||||
|
||||
func loadOwnedFile(ctx *gin.Context) (models.FileObject, bool) {
|
||||
actor, ok := auth.FromContext(ctx)
|
||||
if !ok {
|
||||
@@ -57,12 +70,10 @@ func loadOwnedFile(ctx *gin.Context) (models.FileObject, bool) {
|
||||
func newFileResponse(fileObject models.FileObject) FileResponse {
|
||||
return FileResponse{
|
||||
FileID: fileObject.Identity,
|
||||
ObjectKey: fileObject.ObjectKey,
|
||||
URL: impl.StorageService.PublicURL(fileObject.ObjectKey),
|
||||
Filename: fileObject.OriginalName,
|
||||
Size: fileObject.ActualSize,
|
||||
ContentType: fileObject.ContentType,
|
||||
ETag: fileObject.ETag,
|
||||
StorageETag: fileObject.StorageETag,
|
||||
Status: fileObject.Status,
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,6 +1,7 @@
|
||||
package files
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"math"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
@@ -14,6 +15,7 @@ import (
|
||||
fileerrors "git.apinb.com/ops/files/internal/errors"
|
||||
"git.apinb.com/ops/files/internal/impl"
|
||||
"git.apinb.com/ops/files/internal/models"
|
||||
"git.apinb.com/ops/files/internal/storage"
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
@@ -54,30 +56,39 @@ func InitUpload(ctx *gin.Context) {
|
||||
}
|
||||
|
||||
contentType := strings.TrimSpace(request.ContentType)
|
||||
if contentType == "" || utf8.RuneCountInString(contentType) > 255 {
|
||||
if contentType == "" || utf8.RuneCountInString(contentType) > 255 || strings.ContainsAny(contentType, "\r\n") {
|
||||
infra.Response.Error(ctx, fileerrors.ErrInvalidParameter)
|
||||
return
|
||||
}
|
||||
|
||||
fileID := utils.ULID()
|
||||
objectKey := buildObjectKey(namespaceConfig.Prefix, time.Now(), fileID, extension)
|
||||
upload, err := impl.StorageService.PresignPut(ctx.Request.Context(), objectKey, contentType)
|
||||
expiresAt := time.Now().Add(time.Duration(config.Spec.Storage.AccessTTLSeconds) * time.Second)
|
||||
upload, err := impl.StorageService.PrepareUpload(ctx.Request.Context(), storage.UploadRequest{
|
||||
FileID: fileID, ObjectKey: objectKey, ContentType: contentType,
|
||||
ExpectedSize: request.Size, ExpiresAt: expiresAt,
|
||||
})
|
||||
if err != nil {
|
||||
infra.Response.Error(ctx, fileerrors.ErrObjectStorageOperation)
|
||||
if errors.Is(err, storage.ErrInsufficientSpace) {
|
||||
infra.Response.Error(ctx, fileerrors.ErrStorageSpaceLow)
|
||||
} else {
|
||||
infra.Response.Error(ctx, fileerrors.ErrStorageOperation)
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
fileObject := models.FileObject{
|
||||
Identity: fileID,
|
||||
Namespace: namespace,
|
||||
Provider: config.Spec.ObjectStorage.Provider,
|
||||
Bucket: impl.StorageService.Bucket(),
|
||||
Provider: impl.StorageService.Provider(),
|
||||
StorageContainer: impl.StorageService.Container(),
|
||||
ObjectKey: objectKey,
|
||||
OriginalName: filename,
|
||||
Extension: extension,
|
||||
ContentType: contentType,
|
||||
ExpectedSize: request.Size,
|
||||
Status: models.FileStatusPending,
|
||||
Visibility: models.FileVisibilityPrivate,
|
||||
OwnerType: actor.Type,
|
||||
OwnerID: actor.ID,
|
||||
OwnerIdentity: actor.Identity,
|
||||
@@ -90,7 +101,6 @@ func InitUpload(ctx *gin.Context) {
|
||||
|
||||
infra.Response.Success(ctx, InitUploadResponse{
|
||||
FileID: fileID,
|
||||
ObjectKey: objectKey,
|
||||
Upload: upload,
|
||||
})
|
||||
}
|
||||
|
||||
78
internal/logic/files/local_upload.go
Normal file
78
internal/logic/files/local_upload.go
Normal file
@@ -0,0 +1,78 @@
|
||||
package files
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"net/http"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.apinb.com/ops/files/internal/config"
|
||||
"git.apinb.com/ops/files/internal/impl"
|
||||
"git.apinb.com/ops/files/internal/models"
|
||||
"git.apinb.com/ops/files/internal/signing"
|
||||
"git.apinb.com/ops/files/internal/storage"
|
||||
"github.com/gin-gonic/gin"
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
func LocalUpload(ctx *gin.Context) {
|
||||
receiver, ok := impl.StorageService.(storage.UploadReceiver)
|
||||
if !ok || impl.StorageService.Provider() != "local" {
|
||||
ctx.Status(http.StatusNotFound)
|
||||
return
|
||||
}
|
||||
fileID := strings.TrimSpace(ctx.Param("identity"))
|
||||
expiresUnix, err := strconv.ParseInt(ctx.Query("expires"), 10, 64)
|
||||
if err != nil || fileID == "" {
|
||||
ctx.Status(http.StatusForbidden)
|
||||
return
|
||||
}
|
||||
var fileObject models.FileObject
|
||||
if err := impl.DBService.Where("identity = ?", fileID).First(&fileObject).Error; err != nil {
|
||||
ctx.Status(http.StatusForbidden)
|
||||
return
|
||||
}
|
||||
request := storage.UploadRequest{
|
||||
FileID: fileID, ObjectKey: fileObject.ObjectKey, ContentType: fileObject.ContentType,
|
||||
ExpectedSize: fileObject.ExpectedSize, ExpiresAt: time.Unix(expiresUnix, 0),
|
||||
}
|
||||
if !signing.VerifyUpload(config.Spec.Storage.SigningSecret, ctx.Query("signature"), request, time.Now()) ||
|
||||
ctx.GetHeader("Content-Type") != fileObject.ContentType || ctx.Request.ContentLength != fileObject.ExpectedSize {
|
||||
ctx.Status(http.StatusForbidden)
|
||||
return
|
||||
}
|
||||
release, err := receiver.AcquireUpload(ctx.Request.Context(), fileID)
|
||||
if err != nil {
|
||||
if errors.Is(err, storage.ErrUploadLocked) {
|
||||
ctx.Status(http.StatusConflict)
|
||||
} else {
|
||||
ctx.Status(http.StatusInternalServerError)
|
||||
}
|
||||
return
|
||||
}
|
||||
defer release()
|
||||
|
||||
if err := impl.DBService.Where("id = ?", fileObject.ID).First(&fileObject).Error; err != nil {
|
||||
ctx.Status(http.StatusConflict)
|
||||
return
|
||||
}
|
||||
if fileObject.Status != models.FileStatusPending || fileObject.ExpiresAt == nil || !fileObject.ExpiresAt.After(time.Now()) {
|
||||
ctx.Status(http.StatusConflict)
|
||||
return
|
||||
}
|
||||
request.ObjectKey = fileObject.ObjectKey
|
||||
request.ContentType = fileObject.ContentType
|
||||
request.ExpectedSize = fileObject.ExpectedSize
|
||||
if _, err := receiver.ReceiveUpload(ctx.Request.Context(), request, ctx.Request.Body); err != nil {
|
||||
if errors.Is(err, storage.ErrInsufficientSpace) {
|
||||
ctx.Status(http.StatusInsufficientStorage)
|
||||
} else if errors.Is(err, gorm.ErrRecordNotFound) {
|
||||
ctx.Status(http.StatusConflict)
|
||||
} else {
|
||||
ctx.Status(http.StatusInternalServerError)
|
||||
}
|
||||
return
|
||||
}
|
||||
ctx.Status(http.StatusNoContent)
|
||||
}
|
||||
52
internal/logic/files/range.go
Normal file
52
internal/logic/files/range.go
Normal file
@@ -0,0 +1,52 @@
|
||||
package files
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"git.apinb.com/ops/files/internal/storage"
|
||||
)
|
||||
|
||||
func parseByteRange(value string, size int64) (*storage.ByteRange, error) {
|
||||
value = strings.TrimSpace(value)
|
||||
if value == "" {
|
||||
return nil, nil
|
||||
}
|
||||
if size <= 0 || !strings.HasPrefix(value, "bytes=") {
|
||||
return nil, fmt.Errorf("文件读取范围无效")
|
||||
}
|
||||
rangeValue := strings.TrimPrefix(value, "bytes=")
|
||||
if strings.Contains(rangeValue, ",") {
|
||||
return nil, fmt.Errorf("不支持多段文件读取")
|
||||
}
|
||||
parts := strings.SplitN(rangeValue, "-", 2)
|
||||
if len(parts) != 2 {
|
||||
return nil, fmt.Errorf("文件读取范围无效")
|
||||
}
|
||||
if parts[0] == "" {
|
||||
suffixLength, err := strconv.ParseInt(parts[1], 10, 64)
|
||||
if err != nil || suffixLength <= 0 {
|
||||
return nil, fmt.Errorf("文件读取范围无效")
|
||||
}
|
||||
if suffixLength > size {
|
||||
suffixLength = size
|
||||
}
|
||||
return &storage.ByteRange{Start: size - suffixLength, End: size - 1}, nil
|
||||
}
|
||||
start, err := strconv.ParseInt(parts[0], 10, 64)
|
||||
if err != nil || start < 0 || start >= size {
|
||||
return nil, fmt.Errorf("文件读取范围无效")
|
||||
}
|
||||
end := size - 1
|
||||
if parts[1] != "" {
|
||||
end, err = strconv.ParseInt(parts[1], 10, 64)
|
||||
if err != nil || end < start {
|
||||
return nil, fmt.Errorf("文件读取范围无效")
|
||||
}
|
||||
if end >= size {
|
||||
end = size - 1
|
||||
}
|
||||
}
|
||||
return &storage.ByteRange{Start: start, End: end}, nil
|
||||
}
|
||||
59
internal/logic/files/stream.go
Normal file
59
internal/logic/files/stream.go
Normal file
@@ -0,0 +1,59 @@
|
||||
package files
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"mime"
|
||||
"net/http"
|
||||
"strings"
|
||||
|
||||
"git.apinb.com/ops/files/internal/impl"
|
||||
"git.apinb.com/ops/files/internal/models"
|
||||
"git.apinb.com/ops/files/internal/runtimeinfo"
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
func streamFile(ctx *gin.Context, fileObject models.FileObject, disposition string) {
|
||||
byteRange, err := parseByteRange(ctx.GetHeader("Range"), fileObject.ActualSize)
|
||||
if err != nil {
|
||||
ctx.Header("Content-Range", fmt.Sprintf("bytes */%d", fileObject.ActualSize))
|
||||
ctx.JSON(http.StatusRequestedRangeNotSatisfiable, gin.H{"message": err.Error()})
|
||||
return
|
||||
}
|
||||
body, err := impl.StorageService.Open(ctx.Request.Context(), fileObject.ObjectKey, byteRange)
|
||||
if err != nil {
|
||||
runtimeinfo.RecordReadFailure()
|
||||
ctx.JSON(http.StatusNotFound, gin.H{"message": "文件不存在"})
|
||||
return
|
||||
}
|
||||
defer body.Close()
|
||||
|
||||
status := http.StatusOK
|
||||
length := fileObject.ActualSize
|
||||
if byteRange != nil {
|
||||
status = http.StatusPartialContent
|
||||
length = byteRange.Length()
|
||||
ctx.Header("Content-Range", fmt.Sprintf("bytes %d-%d/%d", byteRange.Start, byteRange.End, fileObject.ActualSize))
|
||||
}
|
||||
ctx.Header("Accept-Ranges", "bytes")
|
||||
ctx.Header("Cache-Control", "no-store")
|
||||
if fileObject.StorageETag != "" {
|
||||
ctx.Header("ETag", fmt.Sprintf("\"%s\"", strings.ReplaceAll(fileObject.StorageETag, "\"", "")))
|
||||
}
|
||||
if fileObject.CompletedAt != nil {
|
||||
ctx.Header("Last-Modified", fileObject.CompletedAt.UTC().Format(http.TimeFormat))
|
||||
}
|
||||
contentDisposition := mime.FormatMediaType(disposition, map[string]string{"filename": fileObject.OriginalName})
|
||||
ctx.Header("Content-Disposition", contentDisposition)
|
||||
ctx.DataFromReader(status, length, fileObject.ContentType, body, nil)
|
||||
}
|
||||
|
||||
func normalizeDisposition(value, defaultValue string) string {
|
||||
switch strings.ToLower(strings.TrimSpace(value)) {
|
||||
case "inline":
|
||||
return "inline"
|
||||
case "attachment":
|
||||
return "attachment"
|
||||
default:
|
||||
return defaultValue
|
||||
}
|
||||
}
|
||||
@@ -3,6 +3,7 @@ package files
|
||||
import (
|
||||
"git.apinb.com/ops/files/internal/models"
|
||||
"git.apinb.com/ops/files/internal/storage"
|
||||
"time"
|
||||
)
|
||||
|
||||
type InitUploadRequest struct {
|
||||
@@ -14,17 +15,33 @@ type InitUploadRequest struct {
|
||||
|
||||
type InitUploadResponse struct {
|
||||
FileID string `json:"file_id"`
|
||||
ObjectKey string `json:"object_key"`
|
||||
Upload storage.UploadInstruction `json:"upload"`
|
||||
}
|
||||
|
||||
type FileResponse struct {
|
||||
FileID string `json:"file_id"`
|
||||
ObjectKey string `json:"object_key"`
|
||||
URL string `json:"url"`
|
||||
Filename string `json:"filename"`
|
||||
Size int64 `json:"size"`
|
||||
ContentType string `json:"content_type"`
|
||||
ETag string `json:"etag"`
|
||||
StorageETag string `json:"storage_etag"`
|
||||
Status models.FileStatus `json:"status"`
|
||||
}
|
||||
|
||||
type AccessInstruction struct {
|
||||
FileID string `json:"file_id"`
|
||||
URL string `json:"url"`
|
||||
Filename string `json:"filename"`
|
||||
ContentType string `json:"content_type"`
|
||||
Size int64 `json:"size"`
|
||||
Disposition string `json:"disposition"`
|
||||
ExpiresAt time.Time `json:"expires_at,omitempty"`
|
||||
}
|
||||
|
||||
type BatchAccessRequest struct {
|
||||
FileIDs []string `json:"file_ids" binding:"required"`
|
||||
Disposition string `json:"disposition"`
|
||||
}
|
||||
|
||||
type SetVisibilityRequest struct {
|
||||
Visibility models.FileVisibility `json:"visibility" binding:"required"`
|
||||
}
|
||||
|
||||
39
internal/logic/files/visibility.go
Normal file
39
internal/logic/files/visibility.go
Normal file
@@ -0,0 +1,39 @@
|
||||
package files
|
||||
|
||||
import (
|
||||
"git.apinb.com/bsm-sdk/core/infra"
|
||||
fileerrors "git.apinb.com/ops/files/internal/errors"
|
||||
"git.apinb.com/ops/files/internal/impl"
|
||||
"git.apinb.com/ops/files/internal/models"
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
func SetVisibility(ctx *gin.Context) {
|
||||
fileObject, ok := loadOwnedFile(ctx)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
var request SetVisibilityRequest
|
||||
if err := ctx.ShouldBindJSON(&request); err != nil ||
|
||||
(request.Visibility != models.FileVisibilityPrivate && request.Visibility != models.FileVisibilityPublic) {
|
||||
infra.Response.Error(ctx, fileerrors.ErrInvalidParameter)
|
||||
return
|
||||
}
|
||||
if fileObject.Namespace != "visual" || fileObject.Status != models.FileStatusReady {
|
||||
infra.Response.Error(ctx, fileerrors.ErrVisibilityNotAllowed)
|
||||
return
|
||||
}
|
||||
result := impl.DBService.Model(&models.FileObject{}).
|
||||
Where("id = ? AND status = ?", fileObject.ID, models.FileStatusReady).
|
||||
Update("visibility", request.Visibility)
|
||||
if result.Error != nil {
|
||||
infra.Response.Error(ctx, fileerrors.ErrDatabaseOperation)
|
||||
return
|
||||
}
|
||||
if result.RowsAffected != 1 {
|
||||
infra.Response.Error(ctx, fileerrors.ErrUploadStatusConflict)
|
||||
return
|
||||
}
|
||||
fileObject.Visibility = request.Visibility
|
||||
infra.Response.Success(ctx, newFileResponse(fileObject))
|
||||
}
|
||||
@@ -7,6 +7,7 @@ import (
|
||||
)
|
||||
|
||||
type FileStatus string
|
||||
type FileVisibility string
|
||||
|
||||
const (
|
||||
FileStatusPending FileStatus = "pending"
|
||||
@@ -15,20 +16,26 @@ const (
|
||||
FileStatusExpired FileStatus = "expired"
|
||||
)
|
||||
|
||||
const (
|
||||
FileVisibilityPrivate FileVisibility = "private"
|
||||
FileVisibilityPublic FileVisibility = "public"
|
||||
)
|
||||
|
||||
type FileObject struct {
|
||||
ID uint `gorm:"primaryKey" json:"id"`
|
||||
Identity string `gorm:"size:64;not null;uniqueIndex" json:"identity"`
|
||||
Namespace string `gorm:"size:64;not null" json:"namespace"`
|
||||
Provider string `gorm:"size:32;not null" json:"provider"`
|
||||
Bucket string `gorm:"size:255;not null" json:"bucket"`
|
||||
ObjectKey string `gorm:"size:512;not null;uniqueIndex" json:"object_key"`
|
||||
StorageContainer string `gorm:"size:255;not null" json:"-"`
|
||||
ObjectKey string `gorm:"size:512;not null;uniqueIndex" json:"-"`
|
||||
OriginalName string `gorm:"size:255;not null" json:"original_name"`
|
||||
Extension string `gorm:"size:32" json:"extension"`
|
||||
ContentType string `gorm:"size:255" json:"content_type"`
|
||||
ExpectedSize int64 `gorm:"not null" json:"expected_size"`
|
||||
ActualSize int64 `json:"actual_size"`
|
||||
ETag string `gorm:"size:255" json:"etag"`
|
||||
Status FileStatus `gorm:"size:16;not null;index:idx_files_object_status_expires_at,priority:1;index:idx_files_object_status_delete_lease,priority:1" json:"status"`
|
||||
StorageETag string `gorm:"column:storage_etag;size:255" json:"storage_etag"`
|
||||
Visibility FileVisibility `gorm:"size:16;not null;default:private;index:idx_files_object_visibility_status,priority:1" json:"visibility"`
|
||||
Status FileStatus `gorm:"size:16;not null;index:idx_files_object_status_expires_at,priority:1;index:idx_files_object_status_delete_lease,priority:1;index:idx_files_object_visibility_status,priority:2" json:"status"`
|
||||
OwnerType string `gorm:"size:64;not null;index:idx_files_object_owner_type_identity,priority:1" json:"owner_type"`
|
||||
OwnerID uint `gorm:"not null" json:"owner_id"`
|
||||
OwnerIdentity string `gorm:"size:64;not null;index:idx_files_object_owner_type_identity,priority:2" json:"owner_identity"`
|
||||
|
||||
66
internal/models/schema_version.go
Normal file
66
internal/models/schema_version.go
Normal file
@@ -0,0 +1,66 @@
|
||||
package models
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
const requiredSchemaScript = "0001_baseline.up.sql"
|
||||
|
||||
var requiredSchema = struct {
|
||||
checksum string
|
||||
tables []string
|
||||
}{
|
||||
checksum: "ee621597df0d1c80e98cf38b941c1ac00e62f570223b9818bdab20c4e6f96689",
|
||||
tables: []string{"files_object"},
|
||||
}
|
||||
|
||||
// RequireSchemaVersion 确认当前数据库已经由受控迁移脚本初始化。
|
||||
func RequireSchemaVersion(db *gorm.DB, service string) error {
|
||||
if db == nil {
|
||||
return fmt.Errorf("%s 数据库连接未初始化", service)
|
||||
}
|
||||
|
||||
var count int64
|
||||
err := db.Raw(`
|
||||
SELECT COUNT(*)
|
||||
FROM public.ops_schema_migrations
|
||||
WHERE service = ? AND script_name = ? AND checksum = ? AND status = 'success'`,
|
||||
service, requiredSchemaScript, requiredSchema.checksum).Scan(&count).Error
|
||||
if err != nil {
|
||||
return fmt.Errorf("%s 数据库版本检查失败,请先执行 migrate-all.sh up --service %s: %w", service, service, err)
|
||||
}
|
||||
if count != 1 {
|
||||
return fmt.Errorf("%s 数据库迁移版本不匹配,请执行 migrate-all.sh status --service %s", service, service)
|
||||
}
|
||||
|
||||
for _, table := range requiredSchema.tables {
|
||||
var valid bool
|
||||
err = db.Raw(`
|
||||
SELECT to_regclass(?) IS NOT NULL AND EXISTS (
|
||||
SELECT 1
|
||||
FROM pg_constraint c
|
||||
JOIN pg_class t ON t.oid = c.conrelid
|
||||
JOIN pg_namespace n ON n.oid = t.relnamespace
|
||||
WHERE n.nspname = 'public' AND t.relname = ? AND c.contype = 'p'
|
||||
)`, "public."+table, table).Scan(&valid).Error
|
||||
if err != nil || !valid {
|
||||
return fmt.Errorf("%s 数据库关键表结构不完整:%s", service, table)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func RequireStorageProvider(db *gorm.DB, provider string) error {
|
||||
var providers []string
|
||||
if err := db.Model(&FileObject{}).Distinct("provider").Pluck("provider", &providers).Error; err != nil {
|
||||
return fmt.Errorf("检查文件存储类型失败: %w", err)
|
||||
}
|
||||
for _, storedProvider := range providers {
|
||||
if storedProvider != provider {
|
||||
return fmt.Errorf("数据库存在 %s 存储的文件,当前配置为 %s,禁止混用存储后端", storedProvider, provider)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -5,6 +5,7 @@ import (
|
||||
|
||||
"git.apinb.com/bsm-sdk/core/middleware"
|
||||
"git.apinb.com/ops/files/internal/auth"
|
||||
"git.apinb.com/ops/files/internal/health"
|
||||
"git.apinb.com/ops/files/internal/logic/files"
|
||||
"git.apinb.com/ops/files/internal/logic/ping"
|
||||
"github.com/gin-gonic/gin"
|
||||
@@ -13,15 +14,28 @@ import (
|
||||
// Register 注册基础服务路由。
|
||||
func Register(srvKey string, engine *gin.Engine) {
|
||||
v1Group := engine.Group(fmt.Sprintf("/%s/v1", srvKey))
|
||||
engine.GET("/ready", health.Ready)
|
||||
v1Group.GET("/ping/hello", ping.Hello)
|
||||
v1Group.GET("/runtime/status", middleware.JwtAuth(true), health.Status)
|
||||
v1Group.PUT("/local/uploads/:identity", files.LocalUpload)
|
||||
v1Group.GET("/access/:identity/content", files.SignedAccess)
|
||||
v1Group.GET("/public/files/:identity/content", files.PublicAccess)
|
||||
|
||||
registerFileRoutes(v1Group.Group("", middleware.JwtAuth(true)))
|
||||
registerFileRoutes(v1Group.Group("/internal", auth.ServiceAuth()))
|
||||
userGroup := v1Group.Group("", middleware.JwtAuth(true))
|
||||
registerFileRoutes(userGroup)
|
||||
userGroup.GET("/files/:identity/access", files.Access)
|
||||
|
||||
internalGroup := v1Group.Group("/internal", auth.ServiceAuth())
|
||||
internalGroup.GET("/runtime/status", health.Status)
|
||||
registerFileRoutes(internalGroup)
|
||||
internalGroup.POST("/files/access", files.BatchAccess)
|
||||
internalGroup.PUT("/files/:identity/visibility", files.SetVisibility)
|
||||
}
|
||||
|
||||
func registerFileRoutes(group *gin.RouterGroup) {
|
||||
group.POST("/uploads/init", files.InitUpload)
|
||||
group.POST("/uploads/:identity/complete", files.CompleteUpload)
|
||||
group.GET("/files/:identity", files.Detail)
|
||||
group.GET("/files/:identity/content", files.Download)
|
||||
group.DELETE("/files/:identity", files.Delete)
|
||||
}
|
||||
|
||||
67
internal/runtimeinfo/storage.go
Normal file
67
internal/runtimeinfo/storage.go
Normal file
@@ -0,0 +1,67 @@
|
||||
package runtimeinfo
|
||||
|
||||
import (
|
||||
"sync/atomic"
|
||||
"time"
|
||||
)
|
||||
|
||||
type Activity struct {
|
||||
UploadSuccess uint64 `json:"upload_success"`
|
||||
UploadFailure uint64 `json:"upload_failure"`
|
||||
ReadFailure uint64 `json:"read_failure"`
|
||||
DeleteSuccess uint64 `json:"delete_success"`
|
||||
DeleteFailure uint64 `json:"delete_failure"`
|
||||
LastUploadAt time.Time `json:"last_upload_at,omitempty"`
|
||||
LastReadFailAt time.Time `json:"last_read_failure_at,omitempty"`
|
||||
LastDeleteAt time.Time `json:"last_delete_at,omitempty"`
|
||||
}
|
||||
|
||||
var counters struct {
|
||||
uploadSuccess atomic.Uint64
|
||||
uploadFailure atomic.Uint64
|
||||
readFailure atomic.Uint64
|
||||
deleteSuccess atomic.Uint64
|
||||
deleteFailure atomic.Uint64
|
||||
lastUpload atomic.Int64
|
||||
lastReadFail atomic.Int64
|
||||
lastDelete atomic.Int64
|
||||
}
|
||||
|
||||
func RecordUploadSuccess() {
|
||||
counters.uploadSuccess.Add(1)
|
||||
counters.lastUpload.Store(time.Now().UTC().UnixMilli())
|
||||
}
|
||||
|
||||
func RecordUploadFailure() { counters.uploadFailure.Add(1) }
|
||||
|
||||
func RecordReadFailure() {
|
||||
counters.readFailure.Add(1)
|
||||
counters.lastReadFail.Store(time.Now().UTC().UnixMilli())
|
||||
}
|
||||
|
||||
func RecordDeleteSuccess() {
|
||||
counters.deleteSuccess.Add(1)
|
||||
counters.lastDelete.Store(time.Now().UTC().UnixMilli())
|
||||
}
|
||||
|
||||
func RecordDeleteFailure() { counters.deleteFailure.Add(1) }
|
||||
|
||||
func Snapshot() Activity {
|
||||
return Activity{
|
||||
UploadSuccess: counters.uploadSuccess.Load(),
|
||||
UploadFailure: counters.uploadFailure.Load(),
|
||||
ReadFailure: counters.readFailure.Load(),
|
||||
DeleteSuccess: counters.deleteSuccess.Load(),
|
||||
DeleteFailure: counters.deleteFailure.Load(),
|
||||
LastUploadAt: timestamp(counters.lastUpload.Load()),
|
||||
LastReadFailAt: timestamp(counters.lastReadFail.Load()),
|
||||
LastDeleteAt: timestamp(counters.lastDelete.Load()),
|
||||
}
|
||||
}
|
||||
|
||||
func timestamp(milliseconds int64) time.Time {
|
||||
if milliseconds <= 0 {
|
||||
return time.Time{}
|
||||
}
|
||||
return time.UnixMilli(milliseconds).UTC()
|
||||
}
|
||||
64
internal/signing/signature.go
Normal file
64
internal/signing/signature.go
Normal file
@@ -0,0 +1,64 @@
|
||||
package signing
|
||||
|
||||
import (
|
||||
"crypto/hmac"
|
||||
"crypto/sha256"
|
||||
"crypto/subtle"
|
||||
"encoding/hex"
|
||||
"fmt"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.apinb.com/ops/files/internal/storage"
|
||||
)
|
||||
|
||||
func SignUpload(secret string, request storage.UploadRequest) string {
|
||||
canonical := strings.Join([]string{
|
||||
"PUT",
|
||||
request.FileID,
|
||||
strconv.FormatInt(request.ExpiresAt.Unix(), 10),
|
||||
strconv.FormatInt(request.ExpectedSize, 10),
|
||||
request.ContentType,
|
||||
}, "\n")
|
||||
return sign(secret, canonical)
|
||||
}
|
||||
|
||||
func VerifyUpload(secret, signature string, request storage.UploadRequest, now time.Time) bool {
|
||||
return request.ExpiresAt.After(now) && verify(secret, SignUpload(secret, request), signature)
|
||||
}
|
||||
|
||||
func SignAccess(secret, fileID string, expiresAt time.Time, disposition string) string {
|
||||
canonical := strings.Join([]string{"GET", fileID, strconv.FormatInt(expiresAt.Unix(), 10), disposition}, "\n")
|
||||
return sign(secret, canonical)
|
||||
}
|
||||
|
||||
func VerifyAccess(secret, fileID string, expiresAt time.Time, disposition, signature string, now time.Time) bool {
|
||||
return expiresAt.After(now) && verify(secret, SignAccess(secret, fileID, expiresAt, disposition), signature)
|
||||
}
|
||||
|
||||
func sign(secret, canonical string) string {
|
||||
mac := hmac.New(sha256.New, []byte(secret))
|
||||
_, _ = mac.Write([]byte(canonical))
|
||||
return hex.EncodeToString(mac.Sum(nil))
|
||||
}
|
||||
|
||||
func verify(secret, expected, actual string) bool {
|
||||
if strings.TrimSpace(secret) == "" || len(actual) != sha256.Size*2 {
|
||||
return false
|
||||
}
|
||||
expectedBytes, expectedErr := hex.DecodeString(expected)
|
||||
actualBytes, actualErr := hex.DecodeString(actual)
|
||||
if expectedErr != nil || actualErr != nil {
|
||||
return false
|
||||
}
|
||||
return subtle.ConstantTimeCompare(expectedBytes, actualBytes) == 1
|
||||
}
|
||||
|
||||
func UploadNonce(secret string, request storage.UploadRequest) (string, error) {
|
||||
signature := SignUpload(secret, request)
|
||||
if len(signature) < 16 {
|
||||
return "", fmt.Errorf("上传签名长度无效")
|
||||
}
|
||||
return signature[:16], nil
|
||||
}
|
||||
@@ -1,8 +1,6 @@
|
||||
package storage
|
||||
package aliyun
|
||||
|
||||
import (
|
||||
"time"
|
||||
|
||||
"git.apinb.com/ops/files/internal/config"
|
||||
"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
|
||||
"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss/credentials"
|
||||
@@ -11,11 +9,10 @@ import (
|
||||
type Client struct {
|
||||
client *oss.Client
|
||||
bucket string
|
||||
publicBaseURL string
|
||||
presignTTL time.Duration
|
||||
presignTTL int64
|
||||
}
|
||||
|
||||
func New(cfg config.ObjectStorageConf) (*Client, error) {
|
||||
func New(cfg config.AliyunStorageConf, accessTTLSeconds int64) (*Client, error) {
|
||||
ossConfig := oss.LoadDefaultConfig().
|
||||
WithRegion(cfg.Region).
|
||||
WithEndpoint(cfg.Endpoint).
|
||||
@@ -25,7 +22,10 @@ func New(cfg config.ObjectStorageConf) (*Client, error) {
|
||||
return &Client{
|
||||
client: oss.NewClient(ossConfig),
|
||||
bucket: cfg.Bucket,
|
||||
publicBaseURL: cfg.PublicBaseURL,
|
||||
presignTTL: time.Duration(cfg.PresignTTLSeconds) * time.Second,
|
||||
presignTTL: accessTTLSeconds,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (c *Client) Provider() string { return "aliyun" }
|
||||
|
||||
func (c *Client) Container() string { return c.bucket }
|
||||
99
internal/storage/aliyun/object.go
Normal file
99
internal/storage/aliyun/object.go
Normal file
@@ -0,0 +1,99 @@
|
||||
package aliyun
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"strings"
|
||||
|
||||
"git.apinb.com/ops/files/internal/storage"
|
||||
"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
|
||||
)
|
||||
|
||||
// Open 返回对象存储中的只读内容流。
|
||||
func (c *Client) Open(ctx context.Context, objectKey string, byteRange *storage.ByteRange) (io.ReadCloser, error) {
|
||||
if strings.TrimSpace(objectKey) == "" {
|
||||
return nil, fmt.Errorf("对象键不能为空")
|
||||
}
|
||||
request := &oss.GetObjectRequest{
|
||||
Bucket: oss.Ptr(c.bucket),
|
||||
Key: oss.Ptr(objectKey),
|
||||
}
|
||||
if byteRange != nil {
|
||||
rangeHeader := fmt.Sprintf("bytes=%d-%d", byteRange.Start, byteRange.End)
|
||||
request.Range = &rangeHeader
|
||||
}
|
||||
result, err := c.client.GetObject(ctx, request)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("读取对象: %w", err)
|
||||
}
|
||||
if result == nil || result.Body == nil {
|
||||
return nil, fmt.Errorf("对象内容为空")
|
||||
}
|
||||
return result.Body, nil
|
||||
}
|
||||
|
||||
func (c *Client) Stat(ctx context.Context, objectKey string) (storage.ObjectInfo, error) {
|
||||
if strings.TrimSpace(objectKey) == "" {
|
||||
return storage.ObjectInfo{}, fmt.Errorf("对象键不能为空")
|
||||
}
|
||||
|
||||
result, err := c.client.GetObjectMeta(ctx, &oss.GetObjectMetaRequest{
|
||||
Bucket: oss.Ptr(c.bucket),
|
||||
Key: oss.Ptr(objectKey),
|
||||
})
|
||||
if err != nil {
|
||||
var serviceError *oss.ServiceError
|
||||
if errors.As(err, &serviceError) && serviceError.Code == "NoSuchKey" {
|
||||
return storage.ObjectInfo{}, storage.ErrObjectNotFound
|
||||
}
|
||||
return storage.ObjectInfo{}, fmt.Errorf("查询对象元数据: %w", err)
|
||||
}
|
||||
if result == nil {
|
||||
return storage.ObjectInfo{}, fmt.Errorf("对象元数据为空")
|
||||
}
|
||||
if result.ContentLength < 0 {
|
||||
return storage.ObjectInfo{}, fmt.Errorf("对象大小无效: %d", result.ContentLength)
|
||||
}
|
||||
if result.ETag == nil {
|
||||
return storage.ObjectInfo{}, fmt.Errorf("对象 ETag 缺失")
|
||||
}
|
||||
etag := strings.TrimSpace(strings.Trim(strings.TrimSpace(*result.ETag), "\""))
|
||||
if etag == "" {
|
||||
return storage.ObjectInfo{}, fmt.Errorf("对象 ETag 无效")
|
||||
}
|
||||
if result.LastModified == nil || result.LastModified.IsZero() {
|
||||
return storage.ObjectInfo{}, fmt.Errorf("对象最后修改时间缺失")
|
||||
}
|
||||
|
||||
return storage.ObjectInfo{
|
||||
Size: result.ContentLength,
|
||||
StorageETag: etag,
|
||||
LastModified: *result.LastModified,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (c *Client) Delete(ctx context.Context, objectKey string) error {
|
||||
if strings.TrimSpace(objectKey) == "" {
|
||||
return fmt.Errorf("对象键不能为空")
|
||||
}
|
||||
|
||||
_, err := c.client.DeleteObject(ctx, &oss.DeleteObjectRequest{
|
||||
Bucket: oss.Ptr(c.bucket),
|
||||
Key: oss.Ptr(objectKey),
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf("删除对象: %w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Client) Status(ctx context.Context) (storage.RuntimeStatus, error) {
|
||||
_, err := c.client.ListObjectsV2(ctx, &oss.ListObjectsV2Request{Bucket: oss.Ptr(c.bucket), MaxKeys: 1})
|
||||
if err != nil {
|
||||
return storage.RuntimeStatus{Provider: c.Provider(), Container: c.Container()}, fmt.Errorf("检查 OSS 存储状态: %w", err)
|
||||
}
|
||||
return storage.RuntimeStatus{Provider: c.Provider(), Container: c.Container(), Writable: true}, nil
|
||||
}
|
||||
37
internal/storage/aliyun/upload.go
Normal file
37
internal/storage/aliyun/upload.go
Normal file
@@ -0,0 +1,37 @@
|
||||
package aliyun
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.apinb.com/ops/files/internal/storage"
|
||||
"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
|
||||
)
|
||||
|
||||
func (c *Client) PrepareUpload(ctx context.Context, request storage.UploadRequest) (storage.UploadInstruction, error) {
|
||||
if strings.TrimSpace(request.ObjectKey) == "" {
|
||||
return storage.UploadInstruction{}, fmt.Errorf("对象键不能为空")
|
||||
}
|
||||
if strings.TrimSpace(request.ContentType) == "" {
|
||||
return storage.UploadInstruction{}, fmt.Errorf("内容类型不能为空")
|
||||
}
|
||||
|
||||
result, err := c.client.Presign(ctx, &oss.PutObjectRequest{
|
||||
Bucket: oss.Ptr(c.bucket),
|
||||
Key: oss.Ptr(request.ObjectKey),
|
||||
ContentType: oss.Ptr(request.ContentType),
|
||||
ForbidOverwrite: oss.Ptr("true"),
|
||||
}, oss.PresignExpires(time.Duration(c.presignTTL)*time.Second))
|
||||
if err != nil {
|
||||
return storage.UploadInstruction{}, fmt.Errorf("生成上传预签名: %w", err)
|
||||
}
|
||||
|
||||
return storage.UploadInstruction{
|
||||
Method: result.Method,
|
||||
URL: result.URL,
|
||||
Headers: result.SignedHeaders,
|
||||
ExpiresAt: result.Expiration,
|
||||
}, nil
|
||||
}
|
||||
77
internal/storage/backend.go
Normal file
77
internal/storage/backend.go
Normal file
@@ -0,0 +1,77 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"io"
|
||||
"time"
|
||||
)
|
||||
|
||||
var (
|
||||
ErrInsufficientSpace = errors.New("文件存储空间不足")
|
||||
ErrUploadLocked = errors.New("文件正在上传")
|
||||
ErrObjectNotFound = errors.New("文件对象不存在")
|
||||
)
|
||||
|
||||
type UploadRequest struct {
|
||||
FileID string
|
||||
ObjectKey string
|
||||
ContentType string
|
||||
ExpectedSize int64
|
||||
ExpiresAt time.Time
|
||||
}
|
||||
|
||||
type UploadInstruction struct {
|
||||
Method string `json:"method"`
|
||||
URL string `json:"url"`
|
||||
Headers map[string]string `json:"headers"`
|
||||
ExpiresAt time.Time `json:"expires_at"`
|
||||
}
|
||||
|
||||
type ByteRange struct {
|
||||
Start int64
|
||||
End int64
|
||||
}
|
||||
|
||||
func (r ByteRange) Length() int64 {
|
||||
return r.End - r.Start + 1
|
||||
}
|
||||
|
||||
type ObjectInfo struct {
|
||||
Size int64
|
||||
StorageETag string
|
||||
LastModified time.Time
|
||||
}
|
||||
|
||||
type DiskStatus struct {
|
||||
Path string `json:"path"`
|
||||
TotalBytes uint64 `json:"total_bytes"`
|
||||
UsedBytes uint64 `json:"used_bytes"`
|
||||
FreeBytes uint64 `json:"free_bytes"`
|
||||
UsedPercent float64 `json:"used_percent"`
|
||||
MinimumFreeBytes uint64 `json:"minimum_free_bytes"`
|
||||
Level string `json:"level"`
|
||||
}
|
||||
|
||||
type RuntimeStatus struct {
|
||||
Provider string `json:"provider"`
|
||||
Container string `json:"container"`
|
||||
Writable bool `json:"writable"`
|
||||
Disk *DiskStatus `json:"disk,omitempty"`
|
||||
}
|
||||
|
||||
type Backend interface {
|
||||
Provider() string
|
||||
Container() string
|
||||
PrepareUpload(context.Context, UploadRequest) (UploadInstruction, error)
|
||||
Open(context.Context, string, *ByteRange) (io.ReadCloser, error)
|
||||
Stat(context.Context, string) (ObjectInfo, error)
|
||||
Delete(context.Context, string) error
|
||||
Status(context.Context) (RuntimeStatus, error)
|
||||
}
|
||||
|
||||
type UploadReceiver interface {
|
||||
AcquireUpload(context.Context, string) (release func() error, err error)
|
||||
ReceiveUpload(context.Context, UploadRequest, io.Reader) (ObjectInfo, error)
|
||||
DiscardUpload(context.Context, string) error
|
||||
}
|
||||
21
internal/storage/factory/factory.go
Normal file
21
internal/storage/factory/factory.go
Normal file
@@ -0,0 +1,21 @@
|
||||
package factory
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
|
||||
"git.apinb.com/ops/files/internal/config"
|
||||
"git.apinb.com/ops/files/internal/storage"
|
||||
"git.apinb.com/ops/files/internal/storage/aliyun"
|
||||
"git.apinb.com/ops/files/internal/storage/local"
|
||||
)
|
||||
|
||||
func New(storageConfig config.StorageConf) (storage.Backend, error) {
|
||||
switch storageConfig.Provider {
|
||||
case "local":
|
||||
return local.New(storageConfig)
|
||||
case "aliyun":
|
||||
return aliyun.New(storageConfig.Aliyun, storageConfig.AccessTTLSeconds)
|
||||
default:
|
||||
return nil, fmt.Errorf("不支持的文件存储类型: %s", storageConfig.Provider)
|
||||
}
|
||||
}
|
||||
46
internal/storage/local/capacity.go
Normal file
46
internal/storage/local/capacity.go
Normal file
@@ -0,0 +1,46 @@
|
||||
package local
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"math"
|
||||
|
||||
"git.apinb.com/ops/files/internal/storage"
|
||||
"github.com/shirou/gopsutil/v3/disk"
|
||||
)
|
||||
|
||||
func (c *Client) diskStatus() (*storage.DiskStatus, error) {
|
||||
usage, err := disk.Usage(c.rootPath)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("读取本地存储磁盘容量: %w", err)
|
||||
}
|
||||
level := "normal"
|
||||
if usage.Free < c.minimumFreeBytes {
|
||||
level = "critical"
|
||||
} else if c.minimumFreeBytes <= math.MaxUint64/2 && usage.Free < c.minimumFreeBytes*2 {
|
||||
level = "warning"
|
||||
}
|
||||
return &storage.DiskStatus{
|
||||
Path: c.rootPath,
|
||||
TotalBytes: usage.Total,
|
||||
UsedBytes: usage.Used,
|
||||
FreeBytes: usage.Free,
|
||||
UsedPercent: usage.UsedPercent,
|
||||
MinimumFreeBytes: c.minimumFreeBytes,
|
||||
Level: level,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (c *Client) ensureCapacity(expectedSize int64) error {
|
||||
if expectedSize < 0 {
|
||||
return fmt.Errorf("文件大小不能为负数")
|
||||
}
|
||||
diskStatus, err := c.diskStatus()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
requested := uint64(expectedSize)
|
||||
if requested > math.MaxUint64-c.minimumFreeBytes || diskStatus.FreeBytes < requested+c.minimumFreeBytes {
|
||||
return storage.ErrInsufficientSpace
|
||||
}
|
||||
return nil
|
||||
}
|
||||
60
internal/storage/local/client.go
Normal file
60
internal/storage/local/client.go
Normal file
@@ -0,0 +1,60 @@
|
||||
package local
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
|
||||
"git.apinb.com/ops/files/internal/config"
|
||||
)
|
||||
|
||||
const containerName = "local"
|
||||
|
||||
type Client struct {
|
||||
rootPath string
|
||||
objectsPath string
|
||||
stagingPath string
|
||||
locksPath string
|
||||
externalBaseURL string
|
||||
signingSecret string
|
||||
accessTTLSeconds int64
|
||||
minimumFreeBytes uint64
|
||||
}
|
||||
|
||||
func New(storageConfig config.StorageConf) (*Client, error) {
|
||||
rootPath, err := filepath.Abs(storageConfig.Local.RootPath)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("解析本地存储根目录: %w", err)
|
||||
}
|
||||
if err := os.MkdirAll(rootPath, 0o750); err != nil {
|
||||
return nil, fmt.Errorf("创建本地存储根目录: %w", err)
|
||||
}
|
||||
rootPath, err = filepath.EvalSymlinks(rootPath)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("解析本地存储真实路径: %w", err)
|
||||
}
|
||||
client := &Client{
|
||||
rootPath: filepath.Clean(rootPath),
|
||||
objectsPath: filepath.Join(rootPath, "objects"),
|
||||
stagingPath: filepath.Join(rootPath, ".staging"),
|
||||
locksPath: filepath.Join(rootPath, ".locks"),
|
||||
externalBaseURL: strings.TrimRight(storageConfig.ExternalBaseURL, "/"),
|
||||
signingSecret: storageConfig.SigningSecret,
|
||||
accessTTLSeconds: storageConfig.AccessTTLSeconds,
|
||||
minimumFreeBytes: storageConfig.Local.MinimumFreeSpaceMB * 1024 * 1024,
|
||||
}
|
||||
for _, directory := range []string{client.objectsPath, client.stagingPath, client.locksPath} {
|
||||
if err := client.ensureDirectory(directory); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
if _, err := client.Status(nil); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return client, nil
|
||||
}
|
||||
|
||||
func (c *Client) Provider() string { return "local" }
|
||||
|
||||
func (c *Client) Container() string { return containerName }
|
||||
152
internal/storage/local/object.go
Normal file
152
internal/storage/local/object.go
Normal file
@@ -0,0 +1,152 @@
|
||||
package local
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
|
||||
"git.apinb.com/ops/files/internal/storage"
|
||||
)
|
||||
|
||||
func (c *Client) Open(ctx context.Context, objectKey string, byteRange *storage.ByteRange) (io.ReadCloser, error) {
|
||||
path, err := c.resolveObjectPath(objectKey)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
file, err := os.Open(path)
|
||||
if err != nil {
|
||||
if errors.Is(err, os.ErrNotExist) {
|
||||
return nil, storage.ErrObjectNotFound
|
||||
}
|
||||
return nil, fmt.Errorf("打开本地文件: %w", err)
|
||||
}
|
||||
if byteRange == nil {
|
||||
return &contextReadCloser{ctx: ctx, reader: file, closer: file}, nil
|
||||
}
|
||||
info, err := file.Stat()
|
||||
if err != nil {
|
||||
_ = file.Close()
|
||||
return nil, err
|
||||
}
|
||||
if byteRange.Start < 0 || byteRange.End < byteRange.Start || byteRange.End >= info.Size() {
|
||||
_ = file.Close()
|
||||
return nil, fmt.Errorf("文件读取范围无效")
|
||||
}
|
||||
if _, err := file.Seek(byteRange.Start, io.SeekStart); err != nil {
|
||||
_ = file.Close()
|
||||
return nil, err
|
||||
}
|
||||
return &contextReadCloser{ctx: ctx, reader: io.LimitReader(file, byteRange.Length()), closer: file}, nil
|
||||
}
|
||||
|
||||
func (c *Client) Stat(ctx context.Context, objectKey string) (storage.ObjectInfo, error) {
|
||||
path, err := c.resolveObjectPath(objectKey)
|
||||
if err != nil {
|
||||
return storage.ObjectInfo{}, err
|
||||
}
|
||||
file, err := os.Open(path)
|
||||
if err != nil {
|
||||
if errors.Is(err, os.ErrNotExist) {
|
||||
return storage.ObjectInfo{}, storage.ErrObjectNotFound
|
||||
}
|
||||
return storage.ObjectInfo{}, fmt.Errorf("打开本地文件: %w", err)
|
||||
}
|
||||
defer file.Close()
|
||||
info, err := file.Stat()
|
||||
if err != nil {
|
||||
return storage.ObjectInfo{}, err
|
||||
}
|
||||
hasher := sha256.New()
|
||||
if _, err := io.Copy(hasher, &contextReader{ctx: ctx, reader: file}); err != nil {
|
||||
return storage.ObjectInfo{}, fmt.Errorf("计算本地文件摘要: %w", err)
|
||||
}
|
||||
return storage.ObjectInfo{
|
||||
Size: info.Size(),
|
||||
StorageETag: "sha256:" + hex.EncodeToString(hasher.Sum(nil)),
|
||||
LastModified: info.ModTime().UTC(),
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (c *Client) Delete(_ context.Context, objectKey string) error {
|
||||
path, err := c.resolveObjectPath(objectKey)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if err := os.Remove(path); err != nil && !os.IsNotExist(err) {
|
||||
return fmt.Errorf("删除本地文件: %w", err)
|
||||
}
|
||||
c.removeEmptyParents(filepath.Dir(path))
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Client) removeEmptyParents(directory string) {
|
||||
boundary := filepath.Clean(c.objectsPath)
|
||||
for current := filepath.Clean(directory); current != boundary; current = filepath.Dir(current) {
|
||||
relative, err := filepath.Rel(boundary, current)
|
||||
if err != nil || relative == ".." || strings.HasPrefix(relative, ".."+string(filepath.Separator)) {
|
||||
return
|
||||
}
|
||||
if err := os.Remove(current); err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Client) Status(_ context.Context) (storage.RuntimeStatus, error) {
|
||||
diskStatus, err := c.diskStatus()
|
||||
if err != nil {
|
||||
return storage.RuntimeStatus{Provider: c.Provider(), Container: c.Container()}, err
|
||||
}
|
||||
probe, err := os.CreateTemp(c.stagingPath, ".write-probe-")
|
||||
if err != nil {
|
||||
return storage.RuntimeStatus{Provider: c.Provider(), Container: c.Container(), Disk: diskStatus}, fmt.Errorf("本地存储目录不可写: %w", err)
|
||||
}
|
||||
probePath := probe.Name()
|
||||
removeProbe := func() { _ = os.Remove(probePath) }
|
||||
defer removeProbe()
|
||||
if err := probe.Chmod(0o640); err != nil {
|
||||
_ = probe.Close()
|
||||
return storage.RuntimeStatus{Provider: c.Provider(), Container: c.Container(), Disk: diskStatus}, err
|
||||
}
|
||||
if _, err := probe.Write([]byte("files-storage-probe")); err != nil {
|
||||
_ = probe.Close()
|
||||
return storage.RuntimeStatus{Provider: c.Provider(), Container: c.Container(), Disk: diskStatus}, err
|
||||
}
|
||||
if err := probe.Sync(); err != nil {
|
||||
_ = probe.Close()
|
||||
return storage.RuntimeStatus{Provider: c.Provider(), Container: c.Container(), Disk: diskStatus}, err
|
||||
}
|
||||
if err := probe.Close(); err != nil {
|
||||
return storage.RuntimeStatus{Provider: c.Provider(), Container: c.Container(), Disk: diskStatus}, err
|
||||
}
|
||||
if err := os.Remove(probePath); err != nil {
|
||||
return storage.RuntimeStatus{Provider: c.Provider(), Container: c.Container(), Disk: diskStatus}, err
|
||||
}
|
||||
return storage.RuntimeStatus{
|
||||
Provider: c.Provider(),
|
||||
Container: c.Container(),
|
||||
Writable: diskStatus.Level != "critical",
|
||||
Disk: diskStatus,
|
||||
}, nil
|
||||
}
|
||||
|
||||
type contextReadCloser struct {
|
||||
ctx context.Context
|
||||
reader io.Reader
|
||||
closer io.Closer
|
||||
}
|
||||
|
||||
func (r *contextReadCloser) Read(buffer []byte) (int, error) {
|
||||
if err := r.ctx.Err(); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return r.reader.Read(buffer)
|
||||
}
|
||||
|
||||
func (r *contextReadCloser) Close() error { return r.closer.Close() }
|
||||
79
internal/storage/local/path.go
Normal file
79
internal/storage/local/path.go
Normal file
@@ -0,0 +1,79 @@
|
||||
package local
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
)
|
||||
|
||||
func (c *Client) resolveObjectPath(objectKey string) (string, error) {
|
||||
if strings.TrimSpace(objectKey) == "" || filepath.IsAbs(objectKey) || strings.Contains(objectKey, "\\") {
|
||||
return "", fmt.Errorf("对象键不是安全的相对路径")
|
||||
}
|
||||
for _, segment := range strings.Split(objectKey, "/") {
|
||||
if segment == "" || segment == "." || segment == ".." {
|
||||
return "", fmt.Errorf("对象键包含无效路径段")
|
||||
}
|
||||
}
|
||||
target := filepath.Join(c.objectsPath, filepath.FromSlash(objectKey))
|
||||
relative, err := filepath.Rel(c.objectsPath, target)
|
||||
if err != nil || relative == ".." || strings.HasPrefix(relative, ".."+string(filepath.Separator)) {
|
||||
return "", fmt.Errorf("对象路径超出存储目录")
|
||||
}
|
||||
if err := c.ensureNoSymlink(filepath.Dir(target), c.objectsPath); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return target, nil
|
||||
}
|
||||
|
||||
func (c *Client) ensureDirectory(directory string) error {
|
||||
if err := os.MkdirAll(directory, 0o750); err != nil {
|
||||
return fmt.Errorf("创建本地存储目录 %s: %w", directory, err)
|
||||
}
|
||||
if err := c.ensureNoSymlink(directory, c.rootPath); err != nil {
|
||||
return err
|
||||
}
|
||||
return os.Chmod(directory, 0o750)
|
||||
}
|
||||
|
||||
func (c *Client) ensureNoSymlink(target, boundary string) error {
|
||||
target = filepath.Clean(target)
|
||||
boundary = filepath.Clean(boundary)
|
||||
relative, err := filepath.Rel(boundary, target)
|
||||
if err != nil || relative == ".." || strings.HasPrefix(relative, ".."+string(filepath.Separator)) {
|
||||
return fmt.Errorf("路径超出本地存储边界")
|
||||
}
|
||||
current := boundary
|
||||
if relative == "." {
|
||||
return rejectSymlink(current)
|
||||
}
|
||||
if err := rejectSymlink(current); err != nil {
|
||||
return err
|
||||
}
|
||||
for _, segment := range strings.Split(relative, string(filepath.Separator)) {
|
||||
current = filepath.Join(current, segment)
|
||||
info, statErr := os.Lstat(current)
|
||||
if os.IsNotExist(statErr) {
|
||||
continue
|
||||
}
|
||||
if statErr != nil {
|
||||
return fmt.Errorf("检查路径 %s: %w", current, statErr)
|
||||
}
|
||||
if info.Mode()&os.ModeSymlink != 0 {
|
||||
return fmt.Errorf("本地存储路径不允许符号链接: %s", current)
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func rejectSymlink(path string) error {
|
||||
info, err := os.Lstat(path)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if info.Mode()&os.ModeSymlink != 0 {
|
||||
return fmt.Errorf("本地存储路径不允许符号链接: %s", path)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
190
internal/storage/local/upload.go
Normal file
190
internal/storage/local/upload.go
Normal file
@@ -0,0 +1,190 @@
|
||||
package local
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"git.apinb.com/ops/files/internal/signing"
|
||||
"git.apinb.com/ops/files/internal/storage"
|
||||
)
|
||||
|
||||
func (c *Client) PrepareUpload(_ context.Context, request storage.UploadRequest) (storage.UploadInstruction, error) {
|
||||
if err := c.ensureCapacity(request.ExpectedSize); err != nil {
|
||||
return storage.UploadInstruction{}, err
|
||||
}
|
||||
signature := signing.SignUpload(c.signingSecret, request)
|
||||
return storage.UploadInstruction{
|
||||
Method: "PUT",
|
||||
URL: fmt.Sprintf("%s/v1/local/uploads/%s?expires=%d&signature=%s",
|
||||
c.externalBaseURL, request.FileID, request.ExpiresAt.Unix(), signature),
|
||||
Headers: map[string]string{
|
||||
"Content-Type": request.ContentType,
|
||||
"Content-Length": fmt.Sprintf("%d", request.ExpectedSize),
|
||||
},
|
||||
ExpiresAt: request.ExpiresAt,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (c *Client) AcquireUpload(ctx context.Context, fileID string) (func() error, error) {
|
||||
if err := ctx.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if !validFileID(fileID) {
|
||||
return nil, fmt.Errorf("文件标识无效")
|
||||
}
|
||||
lockPath := filepath.Join(c.locksPath, fileID+".lock")
|
||||
lockFile, err := os.OpenFile(lockPath, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o640)
|
||||
if err != nil {
|
||||
if os.IsExist(err) {
|
||||
return nil, storage.ErrUploadLocked
|
||||
}
|
||||
return nil, fmt.Errorf("获取上传锁: %w", err)
|
||||
}
|
||||
return func() error {
|
||||
closeErr := lockFile.Close()
|
||||
removeErr := os.Remove(lockPath)
|
||||
if os.IsNotExist(removeErr) {
|
||||
removeErr = nil
|
||||
}
|
||||
if closeErr != nil {
|
||||
return closeErr
|
||||
}
|
||||
return removeErr
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (c *Client) ReceiveUpload(ctx context.Context, request storage.UploadRequest, source io.Reader) (storage.ObjectInfo, error) {
|
||||
if err := c.ensureCapacity(request.ExpectedSize); err != nil {
|
||||
return storage.ObjectInfo{}, err
|
||||
}
|
||||
targetPath, err := c.resolveObjectPath(request.ObjectKey)
|
||||
if err != nil {
|
||||
return storage.ObjectInfo{}, err
|
||||
}
|
||||
if err := c.ensureDirectory(filepath.Dir(targetPath)); err != nil {
|
||||
return storage.ObjectInfo{}, err
|
||||
}
|
||||
if existing, statErr := c.Stat(ctx, request.ObjectKey); statErr == nil {
|
||||
if existing.Size != request.ExpectedSize {
|
||||
return storage.ObjectInfo{}, fmt.Errorf("已存在文件的大小与上传请求不一致")
|
||||
}
|
||||
return existing, nil
|
||||
} else if !errors.Is(statErr, storage.ErrObjectNotFound) {
|
||||
return storage.ObjectInfo{}, statErr
|
||||
}
|
||||
nonce, err := signing.UploadNonce(c.signingSecret, request)
|
||||
if err != nil {
|
||||
return storage.ObjectInfo{}, err
|
||||
}
|
||||
stagingPath := filepath.Join(c.stagingPath, request.FileID+"."+nonce+".part")
|
||||
staging, err := os.OpenFile(stagingPath, os.O_CREATE|os.O_EXCL|os.O_WRONLY, 0o640)
|
||||
if err != nil {
|
||||
return storage.ObjectInfo{}, fmt.Errorf("创建上传暂存文件: %w", err)
|
||||
}
|
||||
committed := false
|
||||
defer func() {
|
||||
_ = staging.Close()
|
||||
if !committed {
|
||||
_ = os.Remove(stagingPath)
|
||||
}
|
||||
}()
|
||||
|
||||
hasher := sha256.New()
|
||||
written, err := io.Copy(io.MultiWriter(staging, hasher), io.LimitReader(&contextReader{ctx: ctx, reader: source}, request.ExpectedSize+1))
|
||||
if err != nil {
|
||||
return storage.ObjectInfo{}, fmt.Errorf("写入上传暂存文件: %w", err)
|
||||
}
|
||||
if written != request.ExpectedSize {
|
||||
return storage.ObjectInfo{}, fmt.Errorf("上传文件大小不匹配: 期望 %d,实际 %d", request.ExpectedSize, written)
|
||||
}
|
||||
if err := staging.Sync(); err != nil {
|
||||
return storage.ObjectInfo{}, fmt.Errorf("同步上传暂存文件: %w", err)
|
||||
}
|
||||
if err := staging.Close(); err != nil {
|
||||
return storage.ObjectInfo{}, fmt.Errorf("关闭上传暂存文件: %w", err)
|
||||
}
|
||||
if err := os.Link(stagingPath, targetPath); err != nil {
|
||||
return storage.ObjectInfo{}, fmt.Errorf("提交上传文件: %w", err)
|
||||
}
|
||||
if err := os.Chmod(targetPath, 0o640); err != nil {
|
||||
_ = os.Remove(targetPath)
|
||||
return storage.ObjectInfo{}, fmt.Errorf("设置上传文件权限: %w", err)
|
||||
}
|
||||
if err := os.Remove(stagingPath); err != nil {
|
||||
_ = os.Remove(targetPath)
|
||||
return storage.ObjectInfo{}, fmt.Errorf("清理上传暂存文件: %w", err)
|
||||
}
|
||||
if err := syncDirectory(filepath.Dir(targetPath)); err != nil {
|
||||
_ = os.Remove(targetPath)
|
||||
return storage.ObjectInfo{}, err
|
||||
}
|
||||
committed = true
|
||||
return storage.ObjectInfo{
|
||||
Size: written,
|
||||
StorageETag: "sha256:" + hex.EncodeToString(hasher.Sum(nil)),
|
||||
LastModified: time.Now().UTC(),
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (c *Client) DiscardUpload(_ context.Context, fileID string) error {
|
||||
if !validFileID(fileID) {
|
||||
return fmt.Errorf("文件标识无效")
|
||||
}
|
||||
entries, err := os.ReadDir(c.stagingPath)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
prefix := fileID + "."
|
||||
for _, entry := range entries {
|
||||
if !entry.IsDir() && strings.HasPrefix(entry.Name(), prefix) && strings.HasSuffix(entry.Name(), ".part") {
|
||||
if err := os.Remove(filepath.Join(c.stagingPath, entry.Name())); err != nil && !os.IsNotExist(err) {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
type contextReader struct {
|
||||
ctx context.Context
|
||||
reader io.Reader
|
||||
}
|
||||
|
||||
func (r *contextReader) Read(buffer []byte) (int, error) {
|
||||
if err := r.ctx.Err(); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return r.reader.Read(buffer)
|
||||
}
|
||||
|
||||
func validFileID(fileID string) bool {
|
||||
if fileID == "" {
|
||||
return false
|
||||
}
|
||||
for _, character := range fileID {
|
||||
if (character < '0' || character > '9') && (character < 'A' || character > 'Z') && (character < 'a' || character > 'z') && character != '-' {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
|
||||
func syncDirectory(path string) error {
|
||||
directory, err := os.Open(path)
|
||||
if err != nil {
|
||||
return fmt.Errorf("打开目录进行同步: %w", err)
|
||||
}
|
||||
defer directory.Close()
|
||||
if err := directory.Sync(); err != nil {
|
||||
return fmt.Errorf("同步文件目录: %w", err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -1,76 +0,0 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
|
||||
)
|
||||
|
||||
type ObjectInfo struct {
|
||||
Size int64
|
||||
ETag string
|
||||
LastModified time.Time
|
||||
}
|
||||
|
||||
func (c *Client) Stat(ctx context.Context, objectKey string) (ObjectInfo, error) {
|
||||
if strings.TrimSpace(objectKey) == "" {
|
||||
return ObjectInfo{}, fmt.Errorf("对象键不能为空")
|
||||
}
|
||||
|
||||
result, err := c.client.GetObjectMeta(ctx, &oss.GetObjectMetaRequest{
|
||||
Bucket: oss.Ptr(c.bucket),
|
||||
Key: oss.Ptr(objectKey),
|
||||
})
|
||||
if err != nil {
|
||||
return ObjectInfo{}, fmt.Errorf("查询对象元数据: %w", err)
|
||||
}
|
||||
if result == nil {
|
||||
return ObjectInfo{}, fmt.Errorf("对象元数据为空")
|
||||
}
|
||||
if result.ContentLength < 0 {
|
||||
return ObjectInfo{}, fmt.Errorf("对象大小无效: %d", result.ContentLength)
|
||||
}
|
||||
if result.ETag == nil {
|
||||
return ObjectInfo{}, fmt.Errorf("对象 ETag 缺失")
|
||||
}
|
||||
etag := strings.TrimSpace(strings.Trim(strings.TrimSpace(*result.ETag), "\""))
|
||||
if etag == "" {
|
||||
return ObjectInfo{}, fmt.Errorf("对象 ETag 无效")
|
||||
}
|
||||
if result.LastModified == nil || result.LastModified.IsZero() {
|
||||
return ObjectInfo{}, fmt.Errorf("对象最后修改时间缺失")
|
||||
}
|
||||
|
||||
return ObjectInfo{
|
||||
Size: result.ContentLength,
|
||||
ETag: etag,
|
||||
LastModified: *result.LastModified,
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (c *Client) Delete(ctx context.Context, objectKey string) error {
|
||||
if strings.TrimSpace(objectKey) == "" {
|
||||
return fmt.Errorf("对象键不能为空")
|
||||
}
|
||||
|
||||
_, err := c.client.DeleteObject(ctx, &oss.DeleteObjectRequest{
|
||||
Bucket: oss.Ptr(c.bucket),
|
||||
Key: oss.Ptr(objectKey),
|
||||
})
|
||||
if err != nil {
|
||||
return fmt.Errorf("删除对象: %w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Client) PublicURL(objectKey string) string {
|
||||
return strings.TrimRight(c.publicBaseURL, "/") + "/" + strings.TrimLeft(objectKey, "/")
|
||||
}
|
||||
|
||||
func (c *Client) Bucket() string {
|
||||
return c.bucket
|
||||
}
|
||||
@@ -1,43 +0,0 @@
|
||||
package storage
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"github.com/aliyun/alibabacloud-oss-go-sdk-v2/oss"
|
||||
)
|
||||
|
||||
type UploadInstruction struct {
|
||||
Method string `json:"method"`
|
||||
URL string `json:"url"`
|
||||
Headers map[string]string `json:"headers"`
|
||||
ExpiresAt time.Time `json:"expires_at"`
|
||||
}
|
||||
|
||||
func (c *Client) PresignPut(ctx context.Context, objectKey, contentType string) (UploadInstruction, error) {
|
||||
if strings.TrimSpace(objectKey) == "" {
|
||||
return UploadInstruction{}, fmt.Errorf("对象键不能为空")
|
||||
}
|
||||
if strings.TrimSpace(contentType) == "" {
|
||||
return UploadInstruction{}, fmt.Errorf("内容类型不能为空")
|
||||
}
|
||||
|
||||
result, err := c.client.Presign(ctx, &oss.PutObjectRequest{
|
||||
Bucket: oss.Ptr(c.bucket),
|
||||
Key: oss.Ptr(objectKey),
|
||||
ContentType: oss.Ptr(contentType),
|
||||
ForbidOverwrite: oss.Ptr("true"),
|
||||
}, oss.PresignExpires(c.presignTTL))
|
||||
if err != nil {
|
||||
return UploadInstruction{}, fmt.Errorf("生成上传预签名: %w", err)
|
||||
}
|
||||
|
||||
return UploadInstruction{
|
||||
Method: result.Method,
|
||||
URL: result.URL,
|
||||
Headers: result.SignedHeaders,
|
||||
ExpiresAt: result.Expiration,
|
||||
}, nil
|
||||
}
|
||||
Reference in New Issue
Block a user