diff --git a/agent/internal/cmd/serve.go b/agent/internal/cmd/serve.go index ca6b9cf..c2f21de 100644 --- a/agent/internal/cmd/serve.go +++ b/agent/internal/cmd/serve.go @@ -1,12 +1,15 @@ package cmd import ( + "context" "fmt" "net" "os" "time" + "github.com/go-co-op/gocron/v2" "gitlink.org.cn/cloudream/storage/agent/internal/http" + "gitlink.org.cn/cloudream/storage/agent/internal/tickevent" "gitlink.org.cn/cloudream/common/pkgs/ioswitch/exec" "gitlink.org.cn/cloudream/common/pkgs/logger" @@ -15,6 +18,7 @@ import ( "gitlink.org.cn/cloudream/storage/agent/internal/config" "gitlink.org.cn/cloudream/storage/agent/internal/task" stgglb "gitlink.org.cn/cloudream/storage/common/globals" + stgmod "gitlink.org.cn/cloudream/storage/common/models" "gitlink.org.cn/cloudream/storage/common/pkgs/accessstat" "gitlink.org.cn/cloudream/storage/common/pkgs/connectivity" "gitlink.org.cn/cloudream/storage/common/pkgs/distlock" @@ -23,6 +27,7 @@ import ( agtrpc "gitlink.org.cn/cloudream/storage/common/pkgs/grpc/agent" "gitlink.org.cn/cloudream/storage/common/pkgs/metacache" "gitlink.org.cn/cloudream/storage/common/pkgs/storage/agtpool" + "gitlink.org.cn/cloudream/storage/common/pkgs/sysevent" "gitlink.org.cn/cloudream/storage/common/pkgs/uploader" "google.golang.org/grpc" @@ -141,6 +146,21 @@ func serve(configPath string) { // 初始化任务管理器 taskMgr := task.NewManager(distlock, &conCol, &dlder, acStat, stgAgts, uploader) + // 初始化系统事件发布器 + evtPub, err := sysevent.NewPublisher(sysevent.ConfigFromMQConfig(config.Cfg().RabbitMQ), &stgmod.SourceHub{ + HubID: hubCfg.Hub.HubID, + HubName: hubCfg.Hub.Name, + }) + if err != nil { + logger.Errorf("new sysevent publisher: %v", err) + os.Exit(1) + } + go servePublisher(evtPub) + + // 初始化定时任务执行器 + sch := setupTickTask(stgAgts, evtPub) + defer sch.Shutdown() + // 启动命令服务器 // TODO 需要设计AgentID持久化机制 agtSvr, err := agtmq.NewServer(cmdsvc.NewService(&taskMgr, stgAgts), config.Cfg().ID, config.Cfg().RabbitMQ) @@ -185,6 +205,63 @@ func downloadHubConfig() coormq.GetHubConfigResp { return *cfgResp } +func servePublisher(evtPub *sysevent.Publisher) { + logger.Info("start serving sysevent publisher") + + ch := evtPub.Start() + +loop: + for { + val, err := ch.Receive().Wait(context.Background()) + if err != nil { + logger.Errorf("sysevent publisher stopped with error: %s", err.Error()) + break + } + + switch val := val.(type) { + case sysevent.PublishError: + logger.Errorf("publishing event: %v", val) + + case sysevent.PublisherExited: + if val.Err != nil { + logger.Errorf("publisher exited with error: %v", val.Err) + } else { + logger.Info("publisher exited") + } + break loop + + case sysevent.OtherError: + logger.Errorf("sysevent: %v", val) + } + } + logger.Info("sysevent publisher stopped") + + // TODO 仅简单结束了程序 + os.Exit(1) +} + +func setupTickTask(agtPool *agtpool.AgentPool, evtPub *sysevent.Publisher) gocron.Scheduler { + sch, err := gocron.NewScheduler() + if err != nil { + logger.Errorf("new cron scheduler: %s", err.Error()) + os.Exit(1) + } + + sch.NewJob(gocron.DailyJob(1, gocron.NewAtTimes( + gocron.NewAtTime(0, 0, 0), + )), gocron.NewTask(tickevent.ReportStorageStats, agtPool, evtPub)) + + sch.NewJob(gocron.DailyJob(1, gocron.NewAtTimes( + gocron.NewAtTime(0, 0, 1), + )), gocron.NewTask(tickevent.ReportHubTransferStats, evtPub)) + + sch.NewJob(gocron.DailyJob(1, gocron.NewAtTimes( + gocron.NewAtTime(0, 0, 2), + )), gocron.NewTask(tickevent.ReportHubStorageTransferStats, evtPub)) + + return sch +} + func serveAgentServer(server *agtmq.Server) { logger.Info("start serving command server") diff --git a/agent/internal/tickevent/report_hub_stats.go b/agent/internal/tickevent/report_hub_stats.go new file mode 100644 index 0000000..7c3c6d1 --- /dev/null +++ b/agent/internal/tickevent/report_hub_stats.go @@ -0,0 +1,58 @@ +package tickevent + +import ( + stgglb "gitlink.org.cn/cloudream/storage/common/globals" + stgmod "gitlink.org.cn/cloudream/storage/common/models" + "gitlink.org.cn/cloudream/storage/common/pkgs/storage/agtpool" + "gitlink.org.cn/cloudream/storage/common/pkgs/sysevent" +) + +func ReportHubTransferStats(evtPub *sysevent.Publisher) { + if stgglb.Stats.HubTransfer == nil { + return + } + + data := stgglb.Stats.HubTransfer.DumpData() + endTime := stgglb.Stats.HubTransfer.Reset() + + for hubID, entry := range data.Entries { + evtPub.Publish(&stgmod.BodyHubTransferStats{ + SourceHubID: *stgglb.Local.HubID, + TargetHubID: hubID, + Send: stgmod.DataTrans{ + TotalTransfer: entry.OutputBytes, + RequestCount: entry.TotalOutput, + FailedRequestCount: entry.TotalInput - entry.SuccessInput, + AvgTransfer: entry.OutputBytes / entry.TotalOutput, + MinTransfer: entry.MinOutputBytes, + MaxTransfer: entry.MaxOutputBytes, + }, + StartTimestamp: data.StartTime, + EndTimestamp: endTime, + }) + } +} + +func ReportHubStorageTransferStats(stgAgts *agtpool.AgentPool, evtPub *sysevent.Publisher) { + if stgglb.Stats.HubStorageTransfer == nil { + return + } + + data := stgglb.Stats.HubStorageTransfer.DumpData() + endTime := stgglb.Stats.HubStorageTransfer.Reset() + + for storageID, stg := range data.Entries { + evtPub.Publish(&stgmod.BodyHubStorageTransferStats{ + HubID: *stgglb.Local.HubID, + StorageID: storageID, + Send: stgmod.DataTrans{ + TotalTransfer: stg.OutputBytes, + RequestCount: stg.TotalOutput, + FailedRequestCount: stg.TotalInput - stg.SuccessInput, + AvgTransfer: stg.OutputBytes / stg.TotalOutput, + }, + StartTimestamp: data.StartTime, + EndTimestamp: endTime, + }) + } +} diff --git a/agent/internal/tickevent/report_storage_stats.go b/agent/internal/tickevent/report_storage_stats.go new file mode 100644 index 0000000..600acf6 --- /dev/null +++ b/agent/internal/tickevent/report_storage_stats.go @@ -0,0 +1,23 @@ +package tickevent + +import ( + stgmod "gitlink.org.cn/cloudream/storage/common/models" + "gitlink.org.cn/cloudream/storage/common/pkgs/storage/agtpool" + "gitlink.org.cn/cloudream/storage/common/pkgs/sysevent" +) + +func ReportStorageStats(agtPool *agtpool.AgentPool, evtPub *sysevent.Publisher) { + stgs := agtPool.GetAllAgents() + for _, stg := range stgs { + shard, err := stg.GetShardStore() + if err != nil { + continue + } + + stats := shard.Stats() + evtPub.Publish(&stgmod.BodyStorageStats{ + StorageID: stg.Info().Storage.StorageID, + DataCount: stats.FileCount, + }) + } +} diff --git a/common/globals/stats.go b/common/globals/stats.go new file mode 100644 index 0000000..87ed604 --- /dev/null +++ b/common/globals/stats.go @@ -0,0 +1,5 @@ +package stgglb + +import "gitlink.org.cn/cloudream/storage/common/pkgs/servicestats" + +var Stats servicestats.StatsHost diff --git a/common/models/datamap.go b/common/models/datamap.go index a363c44..c61b4ac 100644 --- a/common/models/datamap.go +++ b/common/models/datamap.go @@ -12,7 +12,6 @@ import ( type SysEvent struct { Timestamp time.Time `json:"timestamp"` Source SysEventSource `json:"source"` - Category string `json:"category"` Body SysEventBody `json:"body"` } diff --git a/common/pkgs/ioswitch2/agent_worker.go b/common/pkgs/ioswitch2/agent_worker.go index 393819d..757cb86 100644 --- a/common/pkgs/ioswitch2/agent_worker.go +++ b/common/pkgs/ioswitch2/agent_worker.go @@ -7,6 +7,7 @@ import ( "gitlink.org.cn/cloudream/common/pkgs/ioswitch/exec" "gitlink.org.cn/cloudream/common/pkgs/types" cdssdk "gitlink.org.cn/cloudream/common/sdks/storage" + "gitlink.org.cn/cloudream/common/utils/io2" "gitlink.org.cn/cloudream/common/utils/serder" stgglb "gitlink.org.cn/cloudream/storage/common/globals" agtrpc "gitlink.org.cn/cloudream/storage/common/pkgs/grpc/agent" @@ -45,20 +46,34 @@ func (w *AgentWorker) Equals(worker exec.WorkerInfo) bool { } type AgentWorkerClient struct { - cli *agtrpc.PoolClient + hubID cdssdk.HubID + cli *agtrpc.PoolClient } func (c *AgentWorkerClient) ExecutePlan(ctx context.Context, plan exec.Plan) error { return c.cli.ExecuteIOPlan(ctx, plan) } func (c *AgentWorkerClient) SendStream(ctx context.Context, planID exec.PlanID, id exec.VarID, stream io.ReadCloser) error { - return c.cli.SendStream(ctx, planID, id, stream) + return c.cli.SendStream(ctx, planID, id, io2.CounterCloser(stream, func(cnt int64, err error) { + if stgglb.Stats.HubTransfer != nil { + stgglb.Stats.HubTransfer.RecordOutput(c.hubID, cnt, err == nil || err == io.EOF) + } + })) } func (c *AgentWorkerClient) SendVar(ctx context.Context, planID exec.PlanID, id exec.VarID, value exec.VarValue) error { return c.cli.SendVar(ctx, planID, id, value) } func (c *AgentWorkerClient) GetStream(ctx context.Context, planID exec.PlanID, streamID exec.VarID, signalID exec.VarID, signal exec.VarValue) (io.ReadCloser, error) { - return c.cli.GetStream(ctx, planID, streamID, signalID, signal) + str, err := c.cli.GetStream(ctx, planID, streamID, signalID, signal) + if err != nil { + return nil, err + } + + return io2.CounterCloser(str, func(cnt int64, err error) { + if stgglb.Stats.HubTransfer != nil { + stgglb.Stats.HubTransfer.RecordInput(c.hubID, cnt, err == nil || err == io.EOF) + } + }), nil } func (c *AgentWorkerClient) GetVar(ctx context.Context, planID exec.PlanID, varID exec.VarID, signalID exec.VarID, signal exec.VarValue) (exec.VarValue, error) { return c.cli.GetVar(ctx, planID, varID, signalID, signal) diff --git a/common/pkgs/ioswitch2/http_hub_worker.go b/common/pkgs/ioswitch2/http_hub_worker.go index fe4a25d..f145f25 100644 --- a/common/pkgs/ioswitch2/http_hub_worker.go +++ b/common/pkgs/ioswitch2/http_hub_worker.go @@ -8,6 +8,8 @@ import ( "gitlink.org.cn/cloudream/common/pkgs/ioswitch/exec" cdssdk "gitlink.org.cn/cloudream/common/sdks/storage" "gitlink.org.cn/cloudream/common/sdks/storage/cdsapi" + "gitlink.org.cn/cloudream/common/utils/io2" + stgglb "gitlink.org.cn/cloudream/storage/common/globals" ) type HttpHubWorker struct { @@ -44,7 +46,8 @@ func (w *HttpHubWorker) Equals(worker exec.WorkerInfo) bool { } type HttpHubWorkerClient struct { - cli *cdsapi.Client + hubID cdssdk.HubID + cli *cdsapi.Client } func (c *HttpHubWorkerClient) ExecutePlan(ctx context.Context, plan exec.Plan) error { @@ -58,7 +61,11 @@ func (c *HttpHubWorkerClient) SendStream(ctx context.Context, planID exec.PlanID PlanID: planID, VarID: id, }, - Stream: stream, + Stream: io2.CounterCloser(stream, func(cnt int64, err error) { + if stgglb.Stats.HubTransfer != nil { + stgglb.Stats.HubTransfer.RecordOutput(c.hubID, cnt, err == nil || err == io.EOF) + } + }), }) } func (c *HttpHubWorkerClient) SendVar(ctx context.Context, planID exec.PlanID, id exec.VarID, value exec.VarValue) error { @@ -69,12 +76,21 @@ func (c *HttpHubWorkerClient) SendVar(ctx context.Context, planID exec.PlanID, i }) } func (c *HttpHubWorkerClient) GetStream(ctx context.Context, planID exec.PlanID, streamID exec.VarID, signalID exec.VarID, signal exec.VarValue) (io.ReadCloser, error) { - return c.cli.GetStream(cdsapi.GetStreamReq{ + str, err := c.cli.GetStream(cdsapi.GetStreamReq{ PlanID: planID, VarID: streamID, SignalID: signalID, Signal: signal, }) + if err != nil { + return nil, err + } + + return io2.CounterCloser(str, func(cnt int64, err error) { + if stgglb.Stats.HubTransfer != nil { + stgglb.Stats.HubTransfer.RecordInput(c.hubID, cnt, err == nil || err == io.EOF) + } + }), nil } func (c *HttpHubWorkerClient) GetVar(ctx context.Context, planID exec.PlanID, varID exec.VarID, signalID exec.VarID, signal exec.VarValue) (exec.VarValue, error) { resp, err := c.cli.GetVar(cdsapi.GetVarReq{ diff --git a/common/pkgs/servicestats/hub_strorage_transfer.go b/common/pkgs/servicestats/hub_strorage_transfer.go new file mode 100644 index 0000000..4f86b71 --- /dev/null +++ b/common/pkgs/servicestats/hub_strorage_transfer.go @@ -0,0 +1,103 @@ +package servicestats + +import ( + "math" + "sync" + "time" + + cdssdk "gitlink.org.cn/cloudream/common/sdks/storage" + "gitlink.org.cn/cloudream/common/utils/math2" +) + +type HubStorageTransferStats struct { + data HubStorageTransferStatsData + fromHubID cdssdk.HubID + lock *sync.Mutex +} + +type HubStorageTransferStatsData struct { + Entries map[cdssdk.StorageID]*HubStorageTransferStatsEntry + StartTime time.Time +} + +type HubStorageTransferStatsEntry struct { + DestStorageID cdssdk.StorageID + + OutputBytes int64 + MaxOutputBytes int64 + MinOutputBytes int64 + TotalOutput int64 + SuccessOutput int64 + + InputBytes int64 + MaxInputBytes int64 + MinInputBytes int64 + TotalInput int64 + SuccessInput int64 +} + +func (s *HubStorageTransferStats) RecordUpload(dstStorageID cdssdk.StorageID, transferBytes int64, isSuccess bool) { + s.lock.Lock() + defer s.lock.Unlock() + + e := s.data.Entries[dstStorageID] + if e == nil { + e = &HubStorageTransferStatsEntry{ + DestStorageID: dstStorageID, + MinInputBytes: math.MaxInt64, + MinOutputBytes: math.MaxInt64, + } + s.data.Entries[dstStorageID] = e + } + e.OutputBytes += transferBytes + e.MaxOutputBytes = math2.Max(e.MaxOutputBytes, transferBytes) + e.MinOutputBytes = math2.Min(e.MinOutputBytes, transferBytes) + if isSuccess { + e.SuccessOutput++ + } + e.TotalOutput++ +} + +func (s *HubStorageTransferStats) RecordDownload(dstStorageID cdssdk.StorageID, transferBytes int64, isSuccess bool) { + s.lock.Lock() + defer s.lock.Unlock() + + e := s.data.Entries[dstStorageID] + if e == nil { + e = &HubStorageTransferStatsEntry{ + DestStorageID: dstStorageID, + MinInputBytes: math.MaxInt64, + MinOutputBytes: math.MaxInt64, + } + s.data.Entries[dstStorageID] = e + } + e.InputBytes += transferBytes + e.MaxInputBytes = math2.Max(e.MaxInputBytes, transferBytes) + e.MinInputBytes = math2.Min(e.MinInputBytes, transferBytes) + if isSuccess { + e.SuccessInput++ + } +} + +func (s *HubStorageTransferStats) Reset() time.Time { + s.lock.Lock() + defer s.lock.Unlock() + + s.data.Entries = make(map[cdssdk.StorageID]*HubStorageTransferStatsEntry) + s.data.StartTime = time.Now() + return s.data.StartTime +} + +func (s *HubStorageTransferStats) DumpData() HubStorageTransferStatsData { + s.lock.Lock() + defer s.lock.Unlock() + + data := s.data + data.Entries = make(map[cdssdk.StorageID]*HubStorageTransferStatsEntry) + for k, v := range s.data.Entries { + v2 := *v + data.Entries[k] = &v2 + } + + return data +} diff --git a/common/pkgs/servicestats/hub_transfter.go b/common/pkgs/servicestats/hub_transfter.go new file mode 100644 index 0000000..9eeec51 --- /dev/null +++ b/common/pkgs/servicestats/hub_transfter.go @@ -0,0 +1,103 @@ +package servicestats + +import ( + "math" + "sync" + "time" + + cdssdk "gitlink.org.cn/cloudream/common/sdks/storage" + "gitlink.org.cn/cloudream/common/utils/math2" +) + +type HubTransferStats struct { + data HubTransferStatsData + fromHubID cdssdk.HubID + lock *sync.Mutex +} + +type HubTransferStatsData struct { + Entries map[cdssdk.HubID]*HubTransferStatsEntry + StartTime time.Time +} + +type HubTransferStatsEntry struct { + DestHubID cdssdk.HubID + + OutputBytes int64 + MaxOutputBytes int64 + MinOutputBytes int64 + TotalOutput int64 + SuccessOutput int64 + + InputBytes int64 + MaxInputBytes int64 + MinInputBytes int64 + TotalInput int64 + SuccessInput int64 +} + +func (s *HubTransferStats) RecordOutput(dstHubID cdssdk.HubID, transferBytes int64, isSuccess bool) { + s.lock.Lock() + defer s.lock.Unlock() + + e := s.data.Entries[dstHubID] + if e == nil { + e = &HubTransferStatsEntry{ + DestHubID: dstHubID, + MinInputBytes: math.MaxInt64, + MinOutputBytes: math.MaxInt64, + } + s.data.Entries[dstHubID] = e + } + e.OutputBytes += transferBytes + e.MaxOutputBytes = math2.Max(e.MaxOutputBytes, transferBytes) + e.MinOutputBytes = math2.Min(e.MinOutputBytes, transferBytes) + if isSuccess { + e.SuccessOutput++ + } + e.TotalOutput++ +} + +func (s *HubTransferStats) RecordInput(dstHubID cdssdk.HubID, transferBytes int64, isSuccess bool) { + s.lock.Lock() + defer s.lock.Unlock() + + e := s.data.Entries[dstHubID] + if e == nil { + e = &HubTransferStatsEntry{ + DestHubID: dstHubID, + MinInputBytes: math.MaxInt64, + MinOutputBytes: math.MaxInt64, + } + s.data.Entries[dstHubID] = e + } + e.InputBytes += transferBytes + e.MaxInputBytes = math2.Max(e.MaxInputBytes, transferBytes) + e.MinInputBytes = math2.Min(e.MinInputBytes, transferBytes) + if isSuccess { + e.SuccessInput++ + } + e.TotalInput++ +} + +func (s *HubTransferStats) Reset() time.Time { + s.lock.Lock() + defer s.lock.Unlock() + + s.data.StartTime = time.Now() + s.data.Entries = make(map[cdssdk.HubID]*HubTransferStatsEntry) + return s.data.StartTime +} + +func (s *HubTransferStats) DumpData() HubTransferStatsData { + s.lock.Lock() + defer s.lock.Unlock() + + data := s.data + data.Entries = make(map[cdssdk.HubID]*HubTransferStatsEntry) + for k, v := range s.data.Entries { + v2 := *v + data.Entries[k] = &v2 + } + return data +} diff --git a/common/pkgs/servicestats/service_stats.go b/common/pkgs/servicestats/service_stats.go new file mode 100644 index 0000000..51c8333 --- /dev/null +++ b/common/pkgs/servicestats/service_stats.go @@ -0,0 +1,37 @@ +package servicestats + +import ( + "sync" + "time" + + cdssdk "gitlink.org.cn/cloudream/common/sdks/storage" +) + +type StatsHost struct { + // 统计Hub间的传输数据，仅包含当前Hub主动发送或接收的数量。 + HubTransfer *HubTransferStats + // 统计Hub与存储系统间的传输数据，仅包含当前Hub主动发送或接收的数量。 + HubStorageTransfer *HubStorageTransferStats +} + +func (h *StatsHost) SetupHubTransfer(fromHubID cdssdk.HubID) { + h.HubTransfer = &HubTransferStats{ + fromHubID: fromHubID, + lock: &sync.Mutex{}, + data: HubTransferStatsData{ + StartTime: time.Now(), + Entries: make(map[cdssdk.HubID]*HubTransferStatsEntry), + }, + } +} + +func (h *StatsHost) SetupHubStorageTransfer(fromHubID cdssdk.HubID) { + h.HubStorageTransfer = &HubStorageTransferStats{ + fromHubID: fromHubID, + lock: &sync.Mutex{}, + data: HubStorageTransferStatsData{ + StartTime: time.Now(), + Entries: make(map[cdssdk.StorageID]*HubStorageTransferStatsEntry), + }, + } +} diff --git a/common/pkgs/storage/agtpool/pool.go b/common/pkgs/storage/agtpool/pool.go index 0f93632..69ed71e 100644 --- a/common/pkgs/storage/agtpool/pool.go +++ b/common/pkgs/storage/agtpool/pool.go @@ -74,6 +74,18 @@ func (m *AgentPool) GetAgent(stgID cdssdk.StorageID) (types.StorageAgent, error) return stg.Agent, nil } +func (m *AgentPool) GetAllAgents() []types.StorageAgent { + m.lock.Lock() + defer m.lock.Unlock() + + agents := make([]types.StorageAgent, 0, len(m.storages)) + for _, stg := range m.storages { + agents = append(agents, stg.Agent) + } + + return agents +} + // 查找指定Storage的ShardStore组件 func (m *AgentPool) GetShardStore(stgID cdssdk.StorageID) (types.ShardStore, error) { m.lock.Lock() diff --git a/common/pkgs/storage/local/shard_store.go b/common/pkgs/storage/local/shard_store.go index da6b00e..88d8208 100644 --- a/common/pkgs/storage/local/shard_store.go +++ b/common/pkgs/storage/local/shard_store.go @@ -14,6 +14,7 @@ import ( "gitlink.org.cn/cloudream/common/pkgs/logger" cdssdk "gitlink.org.cn/cloudream/common/sdks/storage" "gitlink.org.cn/cloudream/common/utils/io2" + stgglb "gitlink.org.cn/cloudream/storage/common/globals" "gitlink.org.cn/cloudream/storage/common/pkgs/storage/types" ) @@ -133,7 +134,11 @@ func (s *ShardStore) Create(stream io.Reader) (types.FileInfo, error) { return types.FileInfo{}, err } - size, hash, err := s.writeTempFile(file, stream) + counter := io2.Counter(stream) + size, hash, err := s.writeTempFile(file, counter) + if stgglb.Stats.HubStorageTransfer != nil { + stgglb.Stats.HubStorageTransfer.RecordUpload(s.agt.Detail.Storage.StorageID, counter.Count(), err == nil) + } if err != nil { // Name是文件完整路径 s.onCreateFailed(file.Name()) @@ -272,11 +277,17 @@ func (s *ShardStore) Open(opt types.OpenOption) (io.ReadCloser, error) { } } + var ret io.ReadCloser = file + if opt.Length >= 0 { - return io2.Length(file, opt.Length), nil + ret = io2.Length(ret, opt.Length) } - return file, nil + return io2.CounterCloser(ret, func(cnt int64, err error) { + if stgglb.Stats.HubStorageTransfer != nil { + stgglb.Stats.HubStorageTransfer.RecordDownload(s.agt.Detail.Storage.StorageID, cnt, err == nil || err == io.EOF) + } + }), nil } func (s *ShardStore) Info(hash cdssdk.FileHash) (types.FileInfo, error) { diff --git a/common/pkgs/storage/s3/shard_store.go b/common/pkgs/storage/s3/shard_store.go index b53d956..e2e0a5e 100644 --- a/common/pkgs/storage/s3/shard_store.go +++ b/common/pkgs/storage/s3/shard_store.go @@ -17,6 +17,7 @@ import ( cdssdk "gitlink.org.cn/cloudream/common/sdks/storage" "gitlink.org.cn/cloudream/common/utils/io2" "gitlink.org.cn/cloudream/common/utils/os2" + stgglb "gitlink.org.cn/cloudream/storage/common/globals" stgmod "gitlink.org.cn/cloudream/storage/common/models" "gitlink.org.cn/cloudream/storage/common/pkgs/storage/types" ) @@ -178,7 +179,7 @@ func (s *ShardStore) createWithAwsSha256(stream io.Reader) (types.FileInfo, erro key, fileName := s.createTempFile() - counter := io2.NewCounter(stream) + counter := io2.Counter(stream) resp, err := s.cli.PutObject(context.TODO(), &s3.PutObjectInput{ Bucket: aws.String(s.Bucket), @@ -186,6 +187,9 @@ func (s *ShardStore) createWithAwsSha256(stream io.Reader) (types.FileInfo, erro Body: counter, ChecksumAlgorithm: s3types.ChecksumAlgorithmSha256, }) + if stgglb.Stats.HubStorageTransfer != nil { + stgglb.Stats.HubStorageTransfer.RecordUpload(s.Detail.Storage.StorageID, counter.Count(), err == nil) + } if err != nil { log.Warnf("uploading file %v: %v", key, err) @@ -218,13 +222,16 @@ func (s *ShardStore) createWithCalcSha256(stream io.Reader) (types.FileInfo, err key, fileName := s.createTempFile() hashStr := io2.NewReadHasher(sha256.New(), stream) - counter := io2.NewCounter(hashStr) + counter := io2.Counter(hashStr) _, err := s.cli.PutObject(context.TODO(), &s3.PutObjectInput{ Bucket: aws.String(s.Bucket), Key: aws.String(key), Body: counter, }) + if stgglb.Stats.HubStorageTransfer != nil { + stgglb.Stats.HubStorageTransfer.RecordUpload(s.Detail.Storage.StorageID, counter.Count(), err == nil) + } if err != nil { log.Warnf("uploading file %v: %v", key, err) @@ -320,7 +327,11 @@ func (s *ShardStore) Open(opt types.OpenOption) (io.ReadCloser, error) { return nil, err } - return resp.Body, nil + return io2.CounterCloser(resp.Body, func(cnt int64, err error) { + if stgglb.Stats.HubStorageTransfer != nil { + stgglb.Stats.HubStorageTransfer.RecordDownload(s.Detail.Storage.StorageID, cnt, err == nil || err == io.EOF) + } + }), nil } func (s *ShardStore) Info(hash cdssdk.FileHash) (types.FileInfo, error) { diff --git a/common/pkgs/storage/types/shard_store.go b/common/pkgs/storage/types/shard_store.go index 12a5f30..be6e3be 100644 --- a/common/pkgs/storage/types/shard_store.go +++ b/common/pkgs/storage/types/shard_store.go @@ -55,7 +55,7 @@ type Stats struct { // 存储服务状态，如果状态正常，此值应该是StatusOK Status Status // 文件总数 - FileCount int + FileCount int64 // 存储空间总大小 TotalSize int64 // 已使用的存储空间大小，可以超过存储空间总大小 diff --git a/common/pkgs/sysevent/config.go b/common/pkgs/sysevent/config.go index dfe1bf5..9c5b96b 100644 --- a/common/pkgs/sysevent/config.go +++ b/common/pkgs/sysevent/config.go @@ -1,8 +1,19 @@ package sysevent +import "gitlink.org.cn/cloudream/common/pkgs/mq" + type Config struct { Address string `json:"address"` Account string `json:"account"` Password string `json:"password"` VHost string `json:"vhost"` } + +func ConfigFromMQConfig(mqCfg mq.Config) Config { + return Config{ + Address: mqCfg.Address, + Account: mqCfg.Account, + Password: mqCfg.Password, + VHost: mqCfg.VHost, + } +} diff --git a/common/pkgs/sysevent/publisher.go b/common/pkgs/sysevent/publisher.go index 5b42c93..dfb482b 100644 --- a/common/pkgs/sysevent/publisher.go +++ b/common/pkgs/sysevent/publisher.go @@ -2,10 +2,12 @@ package sysevent import ( "fmt" + "time" "github.com/streadway/amqp" "gitlink.org.cn/cloudream/common/pkgs/async" "gitlink.org.cn/cloudream/common/utils/serder" + stgmod "gitlink.org.cn/cloudream/storage/common/models" ) type PublisherEvent interface{} @@ -107,10 +109,13 @@ func (p *Publisher) Start() *async.UnboundChannel[PublisherEvent] { return ch } -// Publish 发布事件，自动补齐时间戳和源信息 -func (p *Publisher) Publish(evt SysEvent) { - // TODO 补齐时间戳和源信息 - p.eventChan.Send(evt) +// Publish 发布事件，会自动补齐必要信息 +func (p *Publisher) Publish(eventBody stgmod.SysEventBody) { + p.eventChan.Send(stgmod.SysEvent{ + Timestamp: time.Now(), + Source: p.thisSource, + Body: eventBody, + }) } // PublishRaw 完全原样发布事件，不补齐任何信息 diff --git a/go.mod b/go.mod index 8c50e1c..9dcf692 100644 --- a/go.mod +++ b/go.mod @@ -11,6 +11,7 @@ require ( github.com/aws/aws-sdk-go-v2/credentials v1.17.47 github.com/aws/aws-sdk-go-v2/service/s3 v1.71.0 github.com/gin-gonic/gin v1.7.7 + github.com/go-co-op/gocron/v2 v2.15.0 github.com/go-sql-driver/mysql v1.8.1 github.com/hashicorp/golang-lru/v2 v2.0.5 github.com/huaweicloud/huaweicloud-sdk-go-obs v3.24.9+incompatible @@ -24,7 +25,7 @@ require ( github.com/spf13/cobra v1.8.0 github.com/spf13/viper v1.19.0 gitlink.org.cn/cloudream/common v0.0.0 - golang.org/x/sync v0.6.0 + golang.org/x/sync v0.7.0 google.golang.org/grpc v1.62.1 google.golang.org/protobuf v1.33.0 gorm.io/gorm v1.25.11 @@ -46,8 +47,10 @@ require ( github.com/hashicorp/hcl v1.0.0 // indirect github.com/jinzhu/inflection v1.0.0 // indirect github.com/jinzhu/now v1.1.5 // indirect + github.com/jonboulle/clockwork v0.4.0 // indirect github.com/magiconair/properties v1.8.7 // indirect github.com/pelletier/go-toml/v2 v2.2.2 // indirect + github.com/robfig/cron/v3 v3.0.1 // indirect github.com/sagikazarmark/locafero v0.4.0 // indirect github.com/sagikazarmark/slog-shim v0.1.0 // indirect github.com/sourcegraph/conc v0.3.0 // indirect @@ -57,7 +60,7 @@ require ( github.com/tjfoc/gmsm v1.4.1 // indirect go.mongodb.org/mongo-driver v1.12.0 // indirect golang.org/x/crypto v0.23.0 // indirect - golang.org/x/exp v0.0.0-20230905200255-921286631fa9 // indirect + golang.org/x/exp v0.0.0-20240613232115-7f521ea00fb8 // indirect golang.org/x/net v0.23.0 // indirect golang.org/x/sys v0.20.0 // indirect golang.org/x/text v0.15.0 // indirect diff --git a/go.sum b/go.sum index 916f9b3..4262e4a 100644 --- a/go.sum +++ b/go.sum @@ -53,6 +53,8 @@ github.com/gin-contrib/sse v0.1.0 h1:Y/yl/+YNO8GZSjAhjMsSuLt29uWRFHdHYUb5lYOV9qE github.com/gin-contrib/sse v0.1.0/go.mod h1:RHrZQHXnP2xjPF+u1gW/2HnVO7nvIa9PG3Gm+fLHvGI= github.com/gin-gonic/gin v1.7.7 h1:3DoBmSbJbZAWqXJC3SLjAPfutPJJRN1U5pALB7EeTTs= github.com/gin-gonic/gin v1.7.7/go.mod h1:axIBovoeJpVj8S3BwE0uPMTeReE4+AfFtqpqaZ1qq1U= +github.com/go-co-op/gocron/v2 v2.15.0 h1:Kpvo71VSihE+RImmpA+3ta5CcMhoRzMGw4dJawrj4zo= +github.com/go-co-op/gocron/v2 v2.15.0/go.mod h1:ZF70ZwEqz0OO4RBXE1sNxnANy/zvwLcattWEFsqpKig= github.com/go-playground/assert/v2 v2.0.1 h1:MsBgLAaY856+nPRTKrp3/OZK38U/wa0CcBYNjji3q3A= github.com/go-playground/assert/v2 v2.0.1/go.mod h1:VDjEfimB/XKnb+ZQfWdccd7VUvScMdVu0Titje2rxJ4= github.com/go-playground/locales v0.13.0 h1:HyWk6mgj5qFqCT5fjGBuRArbVDfE4hi8+e8ceBS/t7Q= @@ -121,6 +123,8 @@ github.com/jinzhu/inflection v1.0.0 h1:K317FqzuhWc8YvSVlFMCCUb36O/S9MCKRDI7QkRKD github.com/jinzhu/inflection v1.0.0/go.mod h1:h+uFLlag+Qp1Va5pdKtLDYj+kHp5pxUVkryuEj+Srlc= github.com/jinzhu/now v1.1.5 h1:/o9tlHleP7gOFmsnYNz3RGnqzefHA47wQpKrrdTIwXQ= github.com/jinzhu/now v1.1.5/go.mod h1:d3SSVoowX0Lcu0IBviAWJpolVfI5UJVZZ7cO71lE/z8= +github.com/jonboulle/clockwork v0.4.0 h1:p4Cf1aMWXnXAUh8lVfewRBx1zaTSYKrKMF2g3ST4RZ4= +github.com/jonboulle/clockwork v0.4.0/go.mod h1:xgRqUGwRcjKCO1vbZUEtSLrqKoPSsUpK7fnezOII0kc= github.com/json-iterator/go v1.1.9/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/uOdHXbAo4= github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM= github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= @@ -170,6 +174,8 @@ github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH github.com/prometheus/client_model v0.0.0-20190812154241-14fe0d1b01d4/go.mod h1:xMI15A0UPsDsEKsMN9yxemIoYk6Tm2C1GtYGdfGttqA= github.com/rivo/uniseg v0.2.0 h1:S1pD9weZBuJdFmowNwbpi7BJ8TNftyUImj/0WQi72jY= github.com/rivo/uniseg v0.2.0/go.mod h1:J6wj4VEh+S6ZtnVlnTBMWIodfgj8LQOQFoIToxlJtxc= +github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs= +github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro= github.com/rogpeppe/go-internal v1.9.0 h1:73kH8U+JUqXU8lRuOHeVHaa/SZPifC7BkcraZVejAe8= github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs= github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= @@ -212,8 +218,9 @@ github.com/stretchr/testify v1.7.4/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= github.com/stretchr/testify v1.8.2/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo= -github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg= github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA= +github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= github.com/subosito/gotenv v1.6.0 h1:9NlTDc1FTs4qu0DDq7AEtTPNw6SVm7uBMsUCUjABIf8= github.com/subosito/gotenv v1.6.0/go.mod h1:Dk4QP5c2W3ibzajGcXpNraDfq2IrhjMIvMSWPKKo0FU= github.com/tjfoc/gmsm v1.4.1 h1:aMe1GlZb+0bLjn+cKTPEvvn9oUEBlJitaZiiBwsbgho= @@ -241,8 +248,8 @@ go.mongodb.org/mongo-driver v1.12.0 h1:aPx33jmn/rQuJXPQLZQ8NtfPQG8CaqgLThFtqRb0P go.mongodb.org/mongo-driver v1.12.0/go.mod h1:AZkxhPnFJUoH7kZlFkVKucV20K387miPfm7oimrSmK0= go.uber.org/atomic v1.10.0 h1:9qC72Qh0+3MqyJbAn8YU5xVq1frD8bn3JtD2oXtafVQ= go.uber.org/atomic v1.10.0/go.mod h1:LUxbIzbOniOlMKjJjyPfpl4v+PKK2cNJn91OQbhoJI0= -go.uber.org/goleak v1.1.12 h1:gZAh5/EyT/HQwlpkCy6wTpqfH9H8Lz8zbm3dZh+OyzA= -go.uber.org/goleak v1.1.12/go.mod h1:cwTWslyiVhfpKIDGSZEM2HlOvcqm+tG4zioyIeLoqMQ= +go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= +go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= go.uber.org/multierr v1.9.0 h1:7fIwc/ZtS0q++VgcfqFDxSBZVv/Xo49/SYnDFupUwlI= go.uber.org/multierr v1.9.0/go.mod h1:X2jQV1h+kxSjClGpnseKVIxpmcjrj7MNnI0bnlfKTVQ= go.uber.org/zap v1.24.0 h1:FiJd5l1UOLj0wCgbSE0rwwXHzEdAZS6hiiSnxJN/D60= @@ -258,8 +265,8 @@ golang.org/x/crypto v0.19.0/go.mod h1:Iy9bg/ha4yyC70EfRS8jz+B6ybOBKMaSxLj6P6oBDf golang.org/x/crypto v0.23.0 h1:dIJU/v2J8Mdglj/8rJ6UUOM3Zc9zLZxVZwwxMooUSAI= golang.org/x/crypto v0.23.0/go.mod h1:CKFgDieR+mRhux2Lsu27y0fO304Db0wZe70UKqHu0v8= golang.org/x/exp v0.0.0-20190121172915-509febef88a4/go.mod h1:CJ0aWSM057203Lf6IL+f9T1iT9GByDxfZKAQTCR3kQA= -golang.org/x/exp v0.0.0-20230905200255-921286631fa9 h1:GoHiUyI/Tp2nVkLI2mCxVkOjsbSXD66ic0XW0js0R9g= -golang.org/x/exp v0.0.0-20230905200255-921286631fa9/go.mod h1:S2oDrQGGwySpoQPVqRShND87VCbxmc6bL1Yd2oYrm6k= +golang.org/x/exp v0.0.0-20240613232115-7f521ea00fb8 h1:yixxcjnhBmY0nkL253HFVIm0JsFHwrHdT3Yh6szTnfY= +golang.org/x/exp v0.0.0-20240613232115-7f521ea00fb8/go.mod h1:jj3sYF3dwk5D+ghuXyeI3r5MFf+NT2An6/9dOA95KSI= golang.org/x/lint v0.0.0-20181026193005-c67002cb31c3/go.mod h1:UVdnD1Gm6xHRNCYTkRU2/jEulfH38KcIWyp/GAMgvoE= golang.org/x/lint v0.0.0-20190227174305-5b3e6a55c961/go.mod h1:wehouNa3lNwaWXcvxsM5YxQ5yQlVC4a0KAMCusXpPoU= golang.org/x/lint v0.0.0-20190313153728-d0100b6bd8b3/go.mod h1:6SW0HCj/g11FgYtHlgUYUwCkIfeOF89ocIRzGO/8vkc= @@ -292,8 +299,8 @@ golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJ golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20220722155255-886fb9371eb4/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.1.0/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= -golang.org/x/sync v0.6.0 h1:5BMeUDZ7vkXGfEr1x9B4bRcTH4lpkTkpdh0T/J+qjbQ= -golang.org/x/sync v0.6.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= +golang.org/x/sync v0.7.0 h1:YsImfSBoP9QPYL0xyKJPq0gcaJdG3rInoqxTWbfQu9M= +golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= golang.org/x/sys v0.0.0-20180830151530-49385e6e1522/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= 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= @@ -370,8 +377,8 @@ google.golang.org/protobuf v1.26.0/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQ google.golang.org/protobuf v1.33.0 h1:uNO2rsAINq/JlFpSdYEKIZ0uKD/R9cpdv0T+yoGwGmI= google.golang.org/protobuf v1.33.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= -gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 h1:YR8cESwS4TdDjEe65xsg0ogRM/Nc3DYOhEAlW+xobZo= -gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= gopkg.in/ini.v1 v1.67.0 h1:Dgnx+6+nfE+IfzjUEISNeydPJh9AXNNsWbGP9KzCsOA= gopkg.in/ini.v1 v1.67.0/go.mod h1:pNLf8WUiyNEtQjuu5G5vTm06TEv9tsIgeAvK8hOrP4k= gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=