HFO4
3 years ago
commit
22bbfe7da1
@ -0,0 +1,62 @@
|
||||
package middleware
|
||||
|
||||
import (
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/cluster"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/serializer"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/slave"
|
||||
"github.com/gin-gonic/gin"
|
||||
"strconv"
|
||||
)
|
||||
|
||||
// MasterMetadata 解析主机节点发来请求的包含主机节点信息的元数据
|
||||
func MasterMetadata() gin.HandlerFunc {
|
||||
return func(c *gin.Context) {
|
||||
c.Set("MasterSiteID", c.GetHeader("X-Site-Id"))
|
||||
c.Set("MasterSiteURL", c.GetHeader("X-Site-Url"))
|
||||
c.Set("MasterVersion", c.GetHeader("X-Cloudreve-Version"))
|
||||
c.Next()
|
||||
}
|
||||
}
|
||||
|
||||
// UseSlaveAria2Instance 从机用于获取对应主机节点的Aria2实例
|
||||
func UseSlaveAria2Instance() gin.HandlerFunc {
|
||||
return func(c *gin.Context) {
|
||||
if siteID, exist := c.Get("MasterSiteID"); exist {
|
||||
// 获取对应主机节点的从机Aria2实例
|
||||
caller, err := slave.DefaultController.GetAria2Instance(siteID.(string))
|
||||
if err != nil {
|
||||
c.JSON(200, serializer.Err(serializer.CodeNotSet, "无法获取 Aria2 实例", err))
|
||||
c.Abort()
|
||||
return
|
||||
}
|
||||
|
||||
c.Set("MasterAria2Instance", caller)
|
||||
c.Next()
|
||||
return
|
||||
}
|
||||
|
||||
c.JSON(200, serializer.ParamErr("未知的主机节点ID", nil))
|
||||
c.Abort()
|
||||
}
|
||||
}
|
||||
|
||||
func SlaveRPCSignRequired() gin.HandlerFunc {
|
||||
return func(c *gin.Context) {
|
||||
nodeID, err := strconv.ParseUint(c.GetHeader("X-Node-Id"), 10, 64)
|
||||
if err != nil {
|
||||
c.JSON(200, serializer.ParamErr("未知的主机节点ID", err))
|
||||
c.Abort()
|
||||
return
|
||||
}
|
||||
|
||||
slaveNode := cluster.Default.GetNodeByID(uint(nodeID))
|
||||
if slaveNode == nil {
|
||||
c.JSON(200, serializer.ParamErr("未知的主机节点ID", err))
|
||||
c.Abort()
|
||||
return
|
||||
}
|
||||
|
||||
SignRequired(slaveNode.MasterAuthInstance())(c)
|
||||
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,91 @@
|
||||
package model
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"github.com/jinzhu/gorm"
|
||||
)
|
||||
|
||||
// Node 从机节点信息模型
|
||||
type Node struct {
|
||||
gorm.Model
|
||||
Status NodeStatus // 节点状态
|
||||
Name string // 节点别名
|
||||
Type ModelType // 节点状态
|
||||
Server string // 服务器地址
|
||||
SlaveKey string `gorm:"type:text"` // 主->从 通信密钥
|
||||
MasterKey string `gorm:"type:text"` // 从->主 通信密钥
|
||||
Aria2Enabled bool // 是否支持用作离线下载节点
|
||||
Aria2Options string `gorm:"type:text"` // 离线下载配置
|
||||
Rank int // 负载均衡权重
|
||||
|
||||
// 数据库忽略字段
|
||||
Aria2OptionsSerialized Aria2Option `gorm:"-"`
|
||||
}
|
||||
|
||||
// Aria2Option 非公有的Aria2配置属性
|
||||
type Aria2Option struct {
|
||||
// RPC 服务器地址
|
||||
Server string `json:"server,omitempty"`
|
||||
// RPC 密钥
|
||||
Token string `json:"token,omitempty"`
|
||||
// 临时下载目录
|
||||
TempPath string `json:"temp_path,omitempty"`
|
||||
// 附加下载配置
|
||||
Options string `json:"options,omitempty"`
|
||||
// 下载监控间隔
|
||||
Interval int `json:"interval,omitempty"`
|
||||
// RPC API 请求超时
|
||||
Timeout int `json:"timeout,omitempty"`
|
||||
}
|
||||
|
||||
type NodeStatus int
|
||||
type ModelType int
|
||||
|
||||
const (
|
||||
NodeActive NodeStatus = iota
|
||||
NodeSuspend
|
||||
)
|
||||
|
||||
const (
|
||||
SlaveNodeType ModelType = iota
|
||||
MasterNodeType
|
||||
)
|
||||
|
||||
// GetNodeByID 用ID获取节点
|
||||
func GetNodeByID(ID interface{}) (Node, error) {
|
||||
var node Node
|
||||
result := DB.First(&node, ID)
|
||||
return node, result.Error
|
||||
}
|
||||
|
||||
// GetNodesByStatus 根据给定状态获取节点
|
||||
func GetNodesByStatus(status ...NodeStatus) ([]Node, error) {
|
||||
var nodes []Node
|
||||
result := DB.Where("status in (?)", status).Find(&nodes)
|
||||
return nodes, result.Error
|
||||
}
|
||||
|
||||
// AfterFind 找到节点后的钩子
|
||||
func (node *Node) AfterFind() (err error) {
|
||||
// 解析离线下载设置到 Aria2OptionsSerialized
|
||||
if node.Aria2Options != "" {
|
||||
err = json.Unmarshal([]byte(node.Aria2Options), &node.Aria2OptionsSerialized)
|
||||
}
|
||||
|
||||
return err
|
||||
}
|
||||
|
||||
// BeforeSave Save策略前的钩子
|
||||
func (node *Node) BeforeSave() (err error) {
|
||||
optionsValue, err := json.Marshal(&node.Aria2OptionsSerialized)
|
||||
node.Aria2Options = string(optionsValue)
|
||||
return err
|
||||
}
|
||||
|
||||
// SetStatus 设置节点启用状态
|
||||
func (node *Node) SetStatus(status NodeStatus) error {
|
||||
node.Status = status
|
||||
return DB.Model(node).Updates(map[string]interface{}{
|
||||
"status": status,
|
||||
}).Error
|
||||
}
|
||||
@ -1,169 +1,65 @@
|
||||
package aria2
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"context"
|
||||
"fmt"
|
||||
"net/url"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
model "github.com/cloudreve/Cloudreve/v3/models"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/aria2/common"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/aria2/monitor"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/aria2/rpc"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/serializer"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/util"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/balancer"
|
||||
)
|
||||
|
||||
// Instance 默认使用的Aria2处理实例
|
||||
var Instance Aria2 = &DummyAria2{}
|
||||
var Instance common.Aria2 = &common.DummyAria2{}
|
||||
|
||||
// LB 获取 Aria2 节点的负载均衡器
|
||||
var LB balancer.Balancer
|
||||
|
||||
// Lock Instance的读写锁
|
||||
var Lock sync.RWMutex
|
||||
|
||||
// EventNotifier 任务状态更新通知处理器
|
||||
var EventNotifier = &Notifier{}
|
||||
|
||||
// Aria2 离线下载处理接口
|
||||
type Aria2 interface {
|
||||
// CreateTask 创建新的任务
|
||||
CreateTask(task *model.Download, options map[string]interface{}) error
|
||||
// 返回状态信息
|
||||
Status(task *model.Download) (rpc.StatusInfo, error)
|
||||
// 取消任务
|
||||
Cancel(task *model.Download) error
|
||||
// 选择要下载的文件
|
||||
Select(task *model.Download, files []int) error
|
||||
}
|
||||
|
||||
const (
|
||||
// URLTask 从URL添加的任务
|
||||
URLTask = iota
|
||||
// TorrentTask 种子任务
|
||||
TorrentTask
|
||||
)
|
||||
|
||||
const (
|
||||
// Ready 准备就绪
|
||||
Ready = iota
|
||||
// Downloading 下载中
|
||||
Downloading
|
||||
// Paused 暂停中
|
||||
Paused
|
||||
// Error 出错
|
||||
Error
|
||||
// Complete 完成
|
||||
Complete
|
||||
// Canceled 取消/停止
|
||||
Canceled
|
||||
// Unknown 未知状态
|
||||
Unknown
|
||||
)
|
||||
|
||||
var (
|
||||
// ErrNotEnabled 功能未开启错误
|
||||
ErrNotEnabled = serializer.NewError(serializer.CodeNoPermissionErr, "离线下载功能未开启", nil)
|
||||
// ErrUserNotFound 未找到下载任务创建者
|
||||
ErrUserNotFound = serializer.NewError(serializer.CodeNotFound, "无法找到任务创建者", nil)
|
||||
)
|
||||
|
||||
// DummyAria2 未开启Aria2功能时使用的默认处理器
|
||||
type DummyAria2 struct {
|
||||
}
|
||||
|
||||
// CreateTask 创建新任务，此处直接返回未开启错误
|
||||
func (instance *DummyAria2) CreateTask(model *model.Download, options map[string]interface{}) error {
|
||||
return ErrNotEnabled
|
||||
}
|
||||
|
||||
// Status 返回未开启错误
|
||||
func (instance *DummyAria2) Status(task *model.Download) (rpc.StatusInfo, error) {
|
||||
return rpc.StatusInfo{}, ErrNotEnabled
|
||||
}
|
||||
|
||||
// Cancel 返回未开启错误
|
||||
func (instance *DummyAria2) Cancel(task *model.Download) error {
|
||||
return ErrNotEnabled
|
||||
}
|
||||
|
||||
// Select 返回未开启错误
|
||||
func (instance *DummyAria2) Select(task *model.Download, files []int) error {
|
||||
return ErrNotEnabled
|
||||
// GetLoadBalancer 返回供Aria2使用的负载均衡器
|
||||
func GetLoadBalancer() balancer.Balancer {
|
||||
Lock.RLock()
|
||||
defer Lock.RUnlock()
|
||||
return LB
|
||||
}
|
||||
|
||||
// Init 初始化
|
||||
func Init(isReload bool) {
|
||||
Lock.Lock()
|
||||
defer Lock.Unlock()
|
||||
|
||||
// 关闭上个初始连接
|
||||
if previousClient, ok := Instance.(*RPCService); ok {
|
||||
if previousClient.Caller != nil {
|
||||
util.Log().Debug("关闭上个 aria2 连接")
|
||||
previousClient.Caller.Close()
|
||||
}
|
||||
}
|
||||
|
||||
options := model.GetSettingByNames("aria2_rpcurl", "aria2_token", "aria2_options")
|
||||
timeout := model.GetIntSetting("aria2_call_timeout", 5)
|
||||
if options["aria2_rpcurl"] == "" {
|
||||
Instance = &DummyAria2{}
|
||||
return
|
||||
}
|
||||
|
||||
util.Log().Info("初始化 aria2 RPC 服务[%s]", options["aria2_rpcurl"])
|
||||
client := &RPCService{}
|
||||
|
||||
// 解析RPC服务地址
|
||||
server, err := url.Parse(options["aria2_rpcurl"])
|
||||
if err != nil {
|
||||
util.Log().Warning("无法解析 aria2 RPC 服务地址，%s", err)
|
||||
Instance = &DummyAria2{}
|
||||
return
|
||||
}
|
||||
server.Path = "/jsonrpc"
|
||||
|
||||
// 加载自定义下载配置
|
||||
var globalOptions map[string]interface{}
|
||||
err = json.Unmarshal([]byte(options["aria2_options"]), &globalOptions)
|
||||
if err != nil {
|
||||
util.Log().Warning("无法解析 aria2 全局配置，%s", err)
|
||||
Instance = &DummyAria2{}
|
||||
return
|
||||
}
|
||||
|
||||
if err := client.Init(server.String(), options["aria2_token"], timeout, globalOptions); err != nil {
|
||||
util.Log().Warning("初始化 aria2 RPC 服务失败，%s", err)
|
||||
Instance = &DummyAria2{}
|
||||
return
|
||||
}
|
||||
|
||||
Instance = client
|
||||
LB = balancer.NewBalancer("RoundRobin")
|
||||
Lock.Unlock()
|
||||
|
||||
if !isReload {
|
||||
// 从数据库中读取未完成任务，创建监控
|
||||
unfinished := model.GetDownloadsByStatus(Ready, Paused, Downloading)
|
||||
unfinished := model.GetDownloadsByStatus(common.Ready, common.Paused, common.Downloading)
|
||||
|
||||
for i := 0; i < len(unfinished); i++ {
|
||||
// 创建任务监控
|
||||
NewMonitor(&unfinished[i])
|
||||
monitor.NewMonitor(&unfinished[i])
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
// getStatus 将给定的状态字符串转换为状态标识数字
|
||||
func getStatus(status string) int {
|
||||
switch status {
|
||||
case "complete":
|
||||
return Complete
|
||||
case "active":
|
||||
return Downloading
|
||||
case "waiting":
|
||||
return Ready
|
||||
case "paused":
|
||||
return Paused
|
||||
case "error":
|
||||
return Error
|
||||
case "removed":
|
||||
return Canceled
|
||||
default:
|
||||
return Unknown
|
||||
// TestRPCConnection 发送测试用的 RPC 请求，测试服务连通性
|
||||
func TestRPCConnection(server, secret string, timeout int) (rpc.VersionInfo, error) {
|
||||
// 解析RPC服务地址
|
||||
rpcServer, err := url.Parse(server)
|
||||
if err != nil {
|
||||
return rpc.VersionInfo{}, fmt.Errorf("cannot parse RPC server: %w", err)
|
||||
}
|
||||
|
||||
rpcServer.Path = "/jsonrpc"
|
||||
caller, err := rpc.New(context.Background(), rpcServer.String(), secret, time.Duration(timeout)*time.Second, nil)
|
||||
if err != nil {
|
||||
return rpc.VersionInfo{}, fmt.Errorf("cannot initialize rpc connection: %w", err)
|
||||
}
|
||||
|
||||
return caller.GetVersion()
|
||||
}
|
||||
|
||||
@ -0,0 +1,114 @@
|
||||
package common
|
||||
|
||||
import (
|
||||
model "github.com/cloudreve/Cloudreve/v3/models"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/aria2/rpc"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/serializer"
|
||||
)
|
||||
|
||||
// Aria2 离线下载处理接口
|
||||
type Aria2 interface {
|
||||
// Init 初始化客户端连接
|
||||
Init() error
|
||||
// CreateTask 创建新的任务
|
||||
CreateTask(task *model.Download, options map[string]interface{}) (string, error)
|
||||
// 返回状态信息
|
||||
Status(task *model.Download) (rpc.StatusInfo, error)
|
||||
// 取消任务
|
||||
Cancel(task *model.Download) error
|
||||
// 选择要下载的文件
|
||||
Select(task *model.Download, files []int) error
|
||||
// 获取离线下载配置
|
||||
GetConfig() model.Aria2Option
|
||||
// 删除临时下载文件
|
||||
DeleteTempFile(*model.Download) error
|
||||
}
|
||||
|
||||
const (
|
||||
// URLTask 从URL添加的任务
|
||||
URLTask = iota
|
||||
// TorrentTask 种子任务
|
||||
TorrentTask
|
||||
)
|
||||
|
||||
const (
|
||||
// Ready 准备就绪
|
||||
Ready = iota
|
||||
// Downloading 下载中
|
||||
Downloading
|
||||
// Paused 暂停中
|
||||
Paused
|
||||
// Error 出错
|
||||
Error
|
||||
// Complete 完成
|
||||
Complete
|
||||
// Canceled 取消/停止
|
||||
Canceled
|
||||
// Unknown 未知状态
|
||||
Unknown
|
||||
)
|
||||
|
||||
var (
|
||||
// ErrNotEnabled 功能未开启错误
|
||||
ErrNotEnabled = serializer.NewError(serializer.CodeNoPermissionErr, "离线下载功能未开启", nil)
|
||||
// ErrUserNotFound 未找到下载任务创建者
|
||||
ErrUserNotFound = serializer.NewError(serializer.CodeNotFound, "无法找到任务创建者", nil)
|
||||
)
|
||||
|
||||
// DummyAria2 未开启Aria2功能时使用的默认处理器
|
||||
type DummyAria2 struct {
|
||||
}
|
||||
|
||||
func (instance *DummyAria2) Init() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// CreateTask 创建新任务，此处直接返回未开启错误
|
||||
func (instance *DummyAria2) CreateTask(model *model.Download, options map[string]interface{}) (string, error) {
|
||||
return "", ErrNotEnabled
|
||||
}
|
||||
|
||||
// Status 返回未开启错误
|
||||
func (instance *DummyAria2) Status(task *model.Download) (rpc.StatusInfo, error) {
|
||||
return rpc.StatusInfo{}, ErrNotEnabled
|
||||
}
|
||||
|
||||
// Cancel 返回未开启错误
|
||||
func (instance *DummyAria2) Cancel(task *model.Download) error {
|
||||
return ErrNotEnabled
|
||||
}
|
||||
|
||||
// Select 返回未开启错误
|
||||
func (instance *DummyAria2) Select(task *model.Download, files []int) error {
|
||||
return ErrNotEnabled
|
||||
}
|
||||
|
||||
// GetConfig 返回空的
|
||||
func (instance *DummyAria2) GetConfig() model.Aria2Option {
|
||||
return model.Aria2Option{}
|
||||
}
|
||||
|
||||
// GetConfig 返回空的
|
||||
func (instance *DummyAria2) DeleteTempFile(src *model.Download) error {
|
||||
return ErrNotEnabled
|
||||
}
|
||||
|
||||
// GetStatus 将给定的状态字符串转换为状态标识数字
|
||||
func GetStatus(status string) int {
|
||||
switch status {
|
||||
case "complete":
|
||||
return Complete
|
||||
case "active":
|
||||
return Downloading
|
||||
case "waiting":
|
||||
return Ready
|
||||
case "paused":
|
||||
return Paused
|
||||
case "error":
|
||||
return Error
|
||||
case "removed":
|
||||
return Canceled
|
||||
default:
|
||||
return Unknown
|
||||
}
|
||||
}
|
||||
@ -1,64 +0,0 @@
|
||||
package aria2
|
||||
|
||||
import (
|
||||
"sync"
|
||||
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/aria2/rpc"
|
||||
)
|
||||
|
||||
// Notifier aria2实践通知处理
|
||||
type Notifier struct {
|
||||
Subscribes sync.Map
|
||||
}
|
||||
|
||||
// Subscribe 订阅事件通知
|
||||
func (notifier *Notifier) Subscribe(target chan StatusEvent, gid string) {
|
||||
notifier.Subscribes.Store(gid, target)
|
||||
}
|
||||
|
||||
// Unsubscribe 取消订阅事件通知
|
||||
func (notifier *Notifier) Unsubscribe(gid string) {
|
||||
notifier.Subscribes.Delete(gid)
|
||||
}
|
||||
|
||||
// Notify 发送通知
|
||||
func (notifier *Notifier) Notify(events []rpc.Event, status int) {
|
||||
for _, event := range events {
|
||||
if target, ok := notifier.Subscribes.Load(event.Gid); ok {
|
||||
target.(chan StatusEvent) <- StatusEvent{
|
||||
GID: event.Gid,
|
||||
Status: status,
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// OnDownloadStart 下载开始
|
||||
func (notifier *Notifier) OnDownloadStart(events []rpc.Event) {
|
||||
notifier.Notify(events, Downloading)
|
||||
}
|
||||
|
||||
// OnDownloadPause 下载暂停
|
||||
func (notifier *Notifier) OnDownloadPause(events []rpc.Event) {
|
||||
notifier.Notify(events, Paused)
|
||||
}
|
||||
|
||||
// OnDownloadStop 下载停止
|
||||
func (notifier *Notifier) OnDownloadStop(events []rpc.Event) {
|
||||
notifier.Notify(events, Canceled)
|
||||
}
|
||||
|
||||
// OnDownloadComplete 下载完成
|
||||
func (notifier *Notifier) OnDownloadComplete(events []rpc.Event) {
|
||||
notifier.Notify(events, Complete)
|
||||
}
|
||||
|
||||
// OnDownloadError 下载出错
|
||||
func (notifier *Notifier) OnDownloadError(events []rpc.Event) {
|
||||
notifier.Notify(events, Error)
|
||||
}
|
||||
|
||||
// OnBtDownloadComplete BT下载完成
|
||||
func (notifier *Notifier) OnBtDownloadComplete(events []rpc.Event) {
|
||||
notifier.Notify(events, Complete)
|
||||
}
|
||||
@ -0,0 +1,15 @@
|
||||
package balancer
|
||||
|
||||
type Balancer interface {
|
||||
NextPeer(nodes interface{}) (error, interface{})
|
||||
}
|
||||
|
||||
// NewBalancer 根据策略标识返回新的负载均衡器
|
||||
func NewBalancer(strategy string) Balancer {
|
||||
switch strategy {
|
||||
case "RoundRobin":
|
||||
return &RoundRobin{}
|
||||
default:
|
||||
return &RoundRobin{}
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,8 @@
|
||||
package balancer
|
||||
|
||||
import "errors"
|
||||
|
||||
var (
|
||||
ErrInputNotSlice = errors.New("Input value is not silice")
|
||||
ErrNoAvaliableNode = errors.New("No nodes avaliable")
|
||||
)
|
||||
@ -0,0 +1,30 @@
|
||||
package balancer
|
||||
|
||||
import (
|
||||
"reflect"
|
||||
"sync/atomic"
|
||||
)
|
||||
|
||||
type RoundRobin struct {
|
||||
current uint64
|
||||
}
|
||||
|
||||
// NextPeer 返回轮盘的下一节点
|
||||
func (r *RoundRobin) NextPeer(nodes interface{}) (error, interface{}) {
|
||||
v := reflect.ValueOf(nodes)
|
||||
if v.Kind() != reflect.Slice {
|
||||
return ErrInputNotSlice, nil
|
||||
}
|
||||
|
||||
if v.Len() == 0 {
|
||||
return ErrNoAvaliableNode, nil
|
||||
}
|
||||
|
||||
next := r.NextIndex(v.Len())
|
||||
return nil, v.Index(next).Interface()
|
||||
}
|
||||
|
||||
// NextIndex 返回下一个节点下标
|
||||
func (r *RoundRobin) NextIndex(total int) int {
|
||||
return int(atomic.AddUint64(&r.current, uint64(1)) % uint64(total))
|
||||
}
|
||||
@ -0,0 +1,8 @@
|
||||
package cluster
|
||||
|
||||
import "errors"
|
||||
|
||||
var (
|
||||
ErrFeatureNotExist = errors.New("No nodes in nodepool match the feature specificed")
|
||||
ErrIlegalPath = errors.New("path out of boundary of setting temp folder")
|
||||
)
|
||||
@ -0,0 +1,60 @@
|
||||
package cluster
|
||||
|
||||
import (
|
||||
model "github.com/cloudreve/Cloudreve/v3/models"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/aria2/common"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/auth"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/serializer"
|
||||
)
|
||||
|
||||
type Node interface {
|
||||
// Init a node from database model
|
||||
Init(node *model.Node)
|
||||
|
||||
// Check if given feature is enabled
|
||||
IsFeatureEnabled(feature string) bool
|
||||
|
||||
// Subscribe node status change to a callback function
|
||||
SubscribeStatusChange(callback func(isActive bool, id uint))
|
||||
|
||||
// Ping the node
|
||||
Ping(req *serializer.NodePingReq) (*serializer.NodePingResp, error)
|
||||
|
||||
// Returns if the node is active
|
||||
IsActive() bool
|
||||
|
||||
// Get instances for aria2 calls
|
||||
GetAria2Instance() common.Aria2
|
||||
|
||||
// Returns unique id of this node
|
||||
ID() uint
|
||||
|
||||
// Kill node and recycle resources
|
||||
Kill()
|
||||
|
||||
// Returns if current node is master node
|
||||
IsMater() bool
|
||||
|
||||
// Get auth instance used to check RPC call from slave to master
|
||||
MasterAuthInstance() auth.Auth
|
||||
|
||||
// Get auth instance used to check RPC call from master to slave
|
||||
SlaveAuthInstance() auth.Auth
|
||||
|
||||
// Get node DB model
|
||||
DBModel() *model.Node
|
||||
}
|
||||
|
||||
// Create new node from DB model
|
||||
func NewNodeFromDBModel(node *model.Node) Node {
|
||||
switch node.Type {
|
||||
case model.SlaveNodeType:
|
||||
slave := &SlaveNode{}
|
||||
slave.Init(node)
|
||||
return slave
|
||||
default:
|
||||
master := &MasterNode{}
|
||||
master.Init(node)
|
||||
return master
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,176 @@
|
||||
package cluster
|
||||
|
||||
import (
|
||||
model "github.com/cloudreve/Cloudreve/v3/models"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/balancer"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/util"
|
||||
"sync"
|
||||
)
|
||||
|
||||
var Default *NodePool
|
||||
|
||||
// 需要分类的节点组
|
||||
var featureGroup = []string{"aria2"}
|
||||
|
||||
// Pool 节点池
|
||||
type Pool interface {
|
||||
// Returns active node selected by given feature and load balancer
|
||||
BalanceNodeByFeature(feature string, lb balancer.Balancer) (error, Node)
|
||||
|
||||
// Returns node by ID
|
||||
GetNodeByID(id uint) Node
|
||||
|
||||
// Add given node into pool. If node existed, refresh node.
|
||||
Add(node *model.Node)
|
||||
|
||||
// Delete and kill node from pool by given node id
|
||||
Delete(id uint)
|
||||
}
|
||||
|
||||
// NodePool 通用节点池
|
||||
type NodePool struct {
|
||||
active map[uint]Node
|
||||
inactive map[uint]Node
|
||||
|
||||
featureMap map[string][]Node
|
||||
|
||||
lock sync.RWMutex
|
||||
}
|
||||
|
||||
// Init 初始化从机节点池
|
||||
func Init() {
|
||||
Default = &NodePool{
|
||||
featureMap: make(map[string][]Node),
|
||||
}
|
||||
if err := Default.initFromDB(); err != nil {
|
||||
util.Log().Warning("节点池初始化失败, %s", err)
|
||||
}
|
||||
}
|
||||
|
||||
func (pool *NodePool) buildIndexMap() {
|
||||
pool.lock.Lock()
|
||||
for _, feature := range featureGroup {
|
||||
pool.featureMap[feature] = make([]Node, 0)
|
||||
}
|
||||
|
||||
for _, v := range pool.active {
|
||||
for _, feature := range featureGroup {
|
||||
if v.IsFeatureEnabled(feature) {
|
||||
pool.featureMap[feature] = append(pool.featureMap[feature], v)
|
||||
}
|
||||
}
|
||||
}
|
||||
pool.lock.Unlock()
|
||||
}
|
||||
|
||||
func (pool *NodePool) GetNodeByID(id uint) Node {
|
||||
pool.lock.RLock()
|
||||
defer pool.lock.RUnlock()
|
||||
|
||||
if node, ok := pool.active[id]; ok {
|
||||
return node
|
||||
}
|
||||
|
||||
return pool.inactive[id]
|
||||
}
|
||||
|
||||
func (pool *NodePool) nodeStatusChange(isActive bool, id uint) {
|
||||
util.Log().Debug("从机节点 [ID=%d] 状态变更 [Active=%t]", id, isActive)
|
||||
pool.lock.Lock()
|
||||
if isActive {
|
||||
node := pool.inactive[id]
|
||||
delete(pool.inactive, id)
|
||||
pool.active[id] = node
|
||||
} else {
|
||||
node := pool.active[id]
|
||||
delete(pool.active, id)
|
||||
pool.inactive[id] = node
|
||||
}
|
||||
pool.lock.Unlock()
|
||||
|
||||
pool.buildIndexMap()
|
||||
}
|
||||
|
||||
func (pool *NodePool) initFromDB() error {
|
||||
nodes, err := model.GetNodesByStatus(model.NodeActive)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
pool.lock.Lock()
|
||||
pool.active = make(map[uint]Node)
|
||||
pool.inactive = make(map[uint]Node)
|
||||
for i := 0; i < len(nodes); i++ {
|
||||
pool.add(&nodes[i])
|
||||
}
|
||||
pool.lock.Unlock()
|
||||
|
||||
pool.buildIndexMap()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (pool *NodePool) add(node *model.Node) {
|
||||
newNode := NewNodeFromDBModel(node)
|
||||
if newNode.IsActive() {
|
||||
pool.active[node.ID] = newNode
|
||||
} else {
|
||||
pool.inactive[node.ID] = newNode
|
||||
}
|
||||
|
||||
// 订阅节点状态变更
|
||||
newNode.SubscribeStatusChange(func(isActive bool, id uint) {
|
||||
pool.nodeStatusChange(isActive, id)
|
||||
})
|
||||
}
|
||||
|
||||
func (pool *NodePool) Add(node *model.Node) {
|
||||
pool.lock.Lock()
|
||||
defer pool.buildIndexMap()
|
||||
defer pool.lock.Unlock()
|
||||
|
||||
if _, ok := pool.active[node.ID]; ok {
|
||||
// TODO: refresh node
|
||||
return
|
||||
}
|
||||
|
||||
if _, ok := pool.inactive[node.ID]; ok {
|
||||
return
|
||||
}
|
||||
|
||||
pool.add(node)
|
||||
}
|
||||
|
||||
func (pool *NodePool) Delete(id uint) {
|
||||
pool.lock.Lock()
|
||||
defer pool.buildIndexMap()
|
||||
defer pool.lock.Unlock()
|
||||
|
||||
if node, ok := pool.active[id]; ok {
|
||||
node.Kill()
|
||||
delete(pool.active, id)
|
||||
return
|
||||
}
|
||||
|
||||
if node, ok := pool.inactive[id]; ok {
|
||||
node.Kill()
|
||||
delete(pool.inactive, id)
|
||||
return
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
// BalanceNodeByFeature 根据 feature 和 LoadBalancer 取出节点
|
||||
func (pool *NodePool) BalanceNodeByFeature(feature string, lb balancer.Balancer) (error, Node) {
|
||||
pool.lock.RLock()
|
||||
defer pool.lock.RUnlock()
|
||||
if nodes, ok := pool.featureMap[feature]; ok {
|
||||
err, res := lb.NextPeer(nodes)
|
||||
if err == nil {
|
||||
return nil, res.(Node)
|
||||
}
|
||||
|
||||
return err, nil
|
||||
}
|
||||
|
||||
return ErrFeatureNotExist, nil
|
||||
}
|
||||
@ -0,0 +1,405 @@
|
||||
package cluster
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
model "github.com/cloudreve/Cloudreve/v3/models"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/aria2/common"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/aria2/rpc"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/auth"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/request"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/serializer"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/util"
|
||||
"io"
|
||||
"net/url"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
type SlaveNode struct {
|
||||
Model *model.Node
|
||||
Active bool
|
||||
|
||||
caller slaveCaller
|
||||
callback func(bool, uint)
|
||||
close chan bool
|
||||
lock sync.RWMutex
|
||||
}
|
||||
|
||||
type slaveCaller struct {
|
||||
parent *SlaveNode
|
||||
Client request.Client
|
||||
}
|
||||
|
||||
// Init 初始化节点
|
||||
func (node *SlaveNode) Init(nodeModel *model.Node) {
|
||||
node.lock.Lock()
|
||||
defer node.lock.Unlock()
|
||||
node.Model = nodeModel
|
||||
|
||||
// Init http request client
|
||||
var endpoint *url.URL
|
||||
if serverURL, err := url.Parse(node.Model.Server); err == nil {
|
||||
var controller *url.URL
|
||||
controller, _ = url.Parse("/api/v3/slave")
|
||||
endpoint = serverURL.ResolveReference(controller)
|
||||
}
|
||||
|
||||
signTTL := model.GetIntSetting("slave_api_timeout", 60)
|
||||
node.caller.Client = request.NewClient(
|
||||
request.WithMasterMeta(),
|
||||
request.WithTimeout(time.Duration(signTTL)*time.Second),
|
||||
request.WithCredential(auth.HMACAuth{SecretKey: []byte(nodeModel.SlaveKey)}, int64(signTTL)),
|
||||
request.WithEndpoint(endpoint.String()),
|
||||
)
|
||||
|
||||
node.caller.parent = node
|
||||
node.Active = true
|
||||
if node.close != nil {
|
||||
node.close <- true
|
||||
}
|
||||
|
||||
go node.StartPingLoop()
|
||||
}
|
||||
|
||||
// IsFeatureEnabled 查询节点的某项功能是否启用
|
||||
func (node *SlaveNode) IsFeatureEnabled(feature string) bool {
|
||||
node.lock.RLock()
|
||||
defer node.lock.RUnlock()
|
||||
|
||||
switch feature {
|
||||
case "aria2":
|
||||
return node.Model.Aria2Enabled
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
// SubscribeStatusChange 订阅节点状态更改
|
||||
func (node *SlaveNode) SubscribeStatusChange(callback func(bool, uint)) {
|
||||
node.lock.Lock()
|
||||
node.callback = callback
|
||||
node.lock.Unlock()
|
||||
}
|
||||
|
||||
// Ping 从机节点，返回从机负载
|
||||
func (node *SlaveNode) Ping(req *serializer.NodePingReq) (*serializer.NodePingResp, error) {
|
||||
reqBodyEncoded, err := json.Marshal(req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
bodyReader := strings.NewReader(string(reqBodyEncoded))
|
||||
|
||||
resp, err := node.caller.Client.Request(
|
||||
"POST",
|
||||
"heartbeat",
|
||||
bodyReader,
|
||||
).CheckHTTPResponse(200).DecodeResponse()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// 处理列取结果
|
||||
if resp.Code != 0 {
|
||||
return nil, serializer.NewErrorFromResponse(resp)
|
||||
}
|
||||
|
||||
var res serializer.NodePingResp
|
||||
|
||||
if resStr, ok := resp.Data.(string); ok {
|
||||
err = json.Unmarshal([]byte(resStr), &res)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
return &res, nil
|
||||
}
|
||||
|
||||
// IsActive 返回节点是否在线
|
||||
func (node *SlaveNode) IsActive() bool {
|
||||
node.lock.RLock()
|
||||
defer node.lock.RUnlock()
|
||||
|
||||
return node.Active
|
||||
}
|
||||
|
||||
// Kill 结束节点内相关循环
|
||||
func (node *SlaveNode) Kill() {
|
||||
node.lock.RLock()
|
||||
defer node.lock.RUnlock()
|
||||
|
||||
if node.close != nil {
|
||||
close(node.close)
|
||||
}
|
||||
}
|
||||
|
||||
// GetAria2Instance 获取从机Aria2实例
|
||||
func (node *SlaveNode) GetAria2Instance() common.Aria2 {
|
||||
node.lock.RLock()
|
||||
defer node.lock.RUnlock()
|
||||
|
||||
if !node.Model.Aria2Enabled {
|
||||
return &common.DummyAria2{}
|
||||
}
|
||||
|
||||
return &node.caller
|
||||
}
|
||||
|
||||
func (node *SlaveNode) ID() uint {
|
||||
node.lock.RLock()
|
||||
defer node.lock.RUnlock()
|
||||
|
||||
return node.Model.ID
|
||||
}
|
||||
|
||||
func (node *SlaveNode) StartPingLoop() {
|
||||
node.lock.Lock()
|
||||
node.close = make(chan bool)
|
||||
node.lock.Unlock()
|
||||
|
||||
tickDuration := time.Duration(model.GetIntSetting("slave_ping_interval", 300)) * time.Second
|
||||
recoverDuration := time.Duration(model.GetIntSetting("slave_recover_interval", 600)) * time.Second
|
||||
pingTicker := time.Duration(0)
|
||||
|
||||
util.Log().Debug("从机节点 [%s] 启动心跳循环", node.Model.Name)
|
||||
retry := 0
|
||||
recoverMode := false
|
||||
isFirstLoop := true
|
||||
|
||||
loop:
|
||||
for {
|
||||
select {
|
||||
case <-time.After(pingTicker):
|
||||
if pingTicker == 0 {
|
||||
pingTicker = tickDuration
|
||||
}
|
||||
|
||||
util.Log().Debug("从机节点 [%s] 发送Ping", node.Model.Name)
|
||||
res, err := node.Ping(node.getHeartbeatContent(isFirstLoop))
|
||||
isFirstLoop = false
|
||||
|
||||
if err != nil {
|
||||
util.Log().Debug("Ping从机节点 [%s] 时发生错误: %s", node.Model.Name, err)
|
||||
retry++
|
||||
if retry >= model.GetIntSetting("slave_node_retry", 3) {
|
||||
util.Log().Debug("从机节点 [%s] Ping 重试已达到最大限制，将从机节点标记为不可用", node.Model.Name)
|
||||
node.changeStatus(false)
|
||||
|
||||
if !recoverMode {
|
||||
// 启动恢复监控循环
|
||||
util.Log().Debug("从机节点 [%s] 进入恢复模式", node.Model.Name)
|
||||
pingTicker = recoverDuration
|
||||
recoverMode = true
|
||||
}
|
||||
}
|
||||
} else {
|
||||
if recoverMode {
|
||||
util.Log().Debug("从机节点 [%s] 复活", node.Model.Name)
|
||||
pingTicker = tickDuration
|
||||
recoverMode = false
|
||||
isFirstLoop = true
|
||||
}
|
||||
|
||||
util.Log().Debug("从机节点 [%s] 状态: %s", node.Model.Name, res)
|
||||
node.changeStatus(true)
|
||||
retry = 0
|
||||
}
|
||||
|
||||
case <-node.close:
|
||||
util.Log().Debug("从机节点 [%s] 收到关闭信号", node.Model.Name)
|
||||
break loop
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (node *SlaveNode) IsMater() bool {
|
||||
return false
|
||||
}
|
||||
|
||||
func (node *SlaveNode) MasterAuthInstance() auth.Auth {
|
||||
node.lock.RLock()
|
||||
defer node.lock.RUnlock()
|
||||
|
||||
return auth.HMACAuth{SecretKey: []byte(node.Model.MasterKey)}
|
||||
}
|
||||
|
||||
func (node *SlaveNode) SlaveAuthInstance() auth.Auth {
|
||||
node.lock.RLock()
|
||||
defer node.lock.RUnlock()
|
||||
|
||||
return auth.HMACAuth{SecretKey: []byte(node.Model.SlaveKey)}
|
||||
}
|
||||
|
||||
func (node *SlaveNode) DBModel() *model.Node {
|
||||
node.lock.RLock()
|
||||
defer node.lock.RUnlock()
|
||||
|
||||
return node.Model
|
||||
}
|
||||
|
||||
// getHeartbeatContent gets serializer.NodePingReq used to send heartbeat to slave
|
||||
func (node *SlaveNode) getHeartbeatContent(isUpdate bool) *serializer.NodePingReq {
|
||||
return &serializer.NodePingReq{
|
||||
SiteURL: model.GetSiteURL().String(),
|
||||
IsUpdate: isUpdate,
|
||||
SiteID: model.GetSettingByName("siteID"),
|
||||
Node: node.Model,
|
||||
CredentialTTL: model.GetIntSetting("slave_api_timeout", 60),
|
||||
}
|
||||
}
|
||||
|
||||
func (node *SlaveNode) changeStatus(isActive bool) {
|
||||
node.lock.RLock()
|
||||
id := node.Model.ID
|
||||
if isActive != node.Active {
|
||||
node.lock.RUnlock()
|
||||
node.lock.Lock()
|
||||
node.Active = isActive
|
||||
node.lock.Unlock()
|
||||
node.callback(isActive, id)
|
||||
} else {
|
||||
node.lock.RUnlock()
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
func (s *slaveCaller) Init() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// SendAria2Call send remote aria2 call to slave node
|
||||
func (s *slaveCaller) SendAria2Call(body *serializer.SlaveAria2Call, scope string) (*serializer.Response, error) {
|
||||
reqReader, err := getAria2RequestBody(body)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return s.Client.Request(
|
||||
"POST",
|
||||
"aria2/"+scope,
|
||||
reqReader,
|
||||
).CheckHTTPResponse(200).DecodeResponse()
|
||||
}
|
||||
|
||||
func (s *slaveCaller) CreateTask(task *model.Download, options map[string]interface{}) (string, error) {
|
||||
s.parent.lock.RLock()
|
||||
defer s.parent.lock.RUnlock()
|
||||
|
||||
req := &serializer.SlaveAria2Call{
|
||||
Task: task,
|
||||
GroupOptions: options,
|
||||
}
|
||||
|
||||
res, err := s.SendAria2Call(req, "task")
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
if res.Code != 0 {
|
||||
return "", serializer.NewErrorFromResponse(res)
|
||||
}
|
||||
|
||||
return res.Data.(string), err
|
||||
}
|
||||
|
||||
func (s *slaveCaller) Status(task *model.Download) (rpc.StatusInfo, error) {
|
||||
s.parent.lock.RLock()
|
||||
defer s.parent.lock.RUnlock()
|
||||
|
||||
req := &serializer.SlaveAria2Call{
|
||||
Task: task,
|
||||
}
|
||||
|
||||
res, err := s.SendAria2Call(req, "status")
|
||||
if err != nil {
|
||||
return rpc.StatusInfo{}, err
|
||||
}
|
||||
|
||||
if res.Code != 0 {
|
||||
return rpc.StatusInfo{}, serializer.NewErrorFromResponse(res)
|
||||
}
|
||||
|
||||
var status rpc.StatusInfo
|
||||
res.GobDecode(&status)
|
||||
|
||||
return status, err
|
||||
}
|
||||
|
||||
func (s *slaveCaller) Cancel(task *model.Download) error {
|
||||
s.parent.lock.RLock()
|
||||
defer s.parent.lock.RUnlock()
|
||||
|
||||
req := &serializer.SlaveAria2Call{
|
||||
Task: task,
|
||||
}
|
||||
|
||||
res, err := s.SendAria2Call(req, "cancel")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if res.Code != 0 {
|
||||
return serializer.NewErrorFromResponse(res)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *slaveCaller) Select(task *model.Download, files []int) error {
|
||||
s.parent.lock.RLock()
|
||||
defer s.parent.lock.RUnlock()
|
||||
|
||||
req := &serializer.SlaveAria2Call{
|
||||
Task: task,
|
||||
Files: files,
|
||||
}
|
||||
|
||||
res, err := s.SendAria2Call(req, "select")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if res.Code != 0 {
|
||||
return serializer.NewErrorFromResponse(res)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *slaveCaller) GetConfig() model.Aria2Option {
|
||||
s.parent.lock.RLock()
|
||||
defer s.parent.lock.RUnlock()
|
||||
|
||||
return s.parent.Model.Aria2OptionsSerialized
|
||||
}
|
||||
|
||||
func (s *slaveCaller) DeleteTempFile(task *model.Download) error {
|
||||
s.parent.lock.RLock()
|
||||
defer s.parent.lock.RUnlock()
|
||||
|
||||
req := &serializer.SlaveAria2Call{
|
||||
Task: task,
|
||||
}
|
||||
|
||||
res, err := s.SendAria2Call(req, "delete")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if res.Code != 0 {
|
||||
return serializer.NewErrorFromResponse(res)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func getAria2RequestBody(body *serializer.SlaveAria2Call) (io.Reader, error) {
|
||||
reqBodyEncoded, err := json.Marshal(body)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
return strings.NewReader(string(reqBodyEncoded)), nil
|
||||
}
|
||||
@ -0,0 +1,25 @@
|
||||
package onedrive
|
||||
|
||||
import "sync"
|
||||
|
||||
// CredentialLock 针对存储策略凭证的锁
|
||||
type CredentialLock interface {
|
||||
Lock(uint)
|
||||
Unlock(uint)
|
||||
}
|
||||
|
||||
var GlobalMutex = mutexMap{}
|
||||
|
||||
type mutexMap struct {
|
||||
locks sync.Map
|
||||
}
|
||||
|
||||
func (m *mutexMap) Lock(id uint) {
|
||||
lock, _ := m.locks.LoadOrStore(id, &sync.Mutex{})
|
||||
lock.(*sync.Mutex).Lock()
|
||||
}
|
||||
|
||||
func (m *mutexMap) Unlock(id uint) {
|
||||
lock, _ := m.locks.LoadOrStore(id, &sync.Mutex{})
|
||||
lock.(*sync.Mutex).Unlock()
|
||||
}
|
||||
@ -0,0 +1,7 @@
|
||||
package masterinslave
|
||||
|
||||
import "errors"
|
||||
|
||||
var (
|
||||
ErrNotImplemented = errors.New("this method of shadowed policy is not implemented")
|
||||
)
|
||||
@ -0,0 +1,56 @@
|
||||
package masterinslave
|
||||
|
||||
import (
|
||||
"context"
|
||||
model "github.com/cloudreve/Cloudreve/v3/models"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/cluster"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/filesystem/driver"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/filesystem/response"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/serializer"
|
||||
"io"
|
||||
"net/url"
|
||||
)
|
||||
|
||||
// Driver 影子存储策略，用于在从机端上传文件
|
||||
type Driver struct {
|
||||
master cluster.Node
|
||||
handler driver.Handler
|
||||
policy *model.Policy
|
||||
}
|
||||
|
||||
// NewDriver 返回新的处理器
|
||||
func NewDriver(master cluster.Node, handler driver.Handler, policy *model.Policy) driver.Handler {
|
||||
return &Driver{
|
||||
master: master,
|
||||
handler: handler,
|
||||
policy: policy,
|
||||
}
|
||||
}
|
||||
|
||||
func (d *Driver) Put(ctx context.Context, file io.ReadCloser, dst string, size uint64) error {
|
||||
return d.handler.Put(ctx, file, dst, size)
|
||||
}
|
||||
|
||||
func (d *Driver) Delete(ctx context.Context, files []string) ([]string, error) {
|
||||
return d.handler.Delete(ctx, files)
|
||||
}
|
||||
|
||||
func (d *Driver) Get(ctx context.Context, path string) (response.RSCloser, error) {
|
||||
return nil, ErrNotImplemented
|
||||
}
|
||||
|
||||
func (d *Driver) Thumb(ctx context.Context, path string) (*response.ContentResponse, error) {
|
||||
return nil, ErrNotImplemented
|
||||
}
|
||||
|
||||
func (d *Driver) Source(ctx context.Context, path string, url url.URL, ttl int64, isDownload bool, speed int) (string, error) {
|
||||
return "", ErrNotImplemented
|
||||
}
|
||||
|
||||
func (d *Driver) Token(ctx context.Context, ttl int64, callbackKey string) (serializer.UploadCredential, error) {
|
||||
return serializer.UploadCredential{}, ErrNotImplemented
|
||||
}
|
||||
|
||||
func (d *Driver) List(ctx context.Context, path string, recursive bool) ([]response.Object, error) {
|
||||
return nil, ErrNotImplemented
|
||||
}
|
||||
@ -0,0 +1,9 @@
|
||||
package slaveinmaster
|
||||
|
||||
import "errors"
|
||||
|
||||
var (
|
||||
ErrNotImplemented = errors.New("this method of shadowed policy is not implemented")
|
||||
ErrSlaveSrcPathNotExist = errors.New("cannot determine source file path in slave node")
|
||||
ErrWaitResultTimeout = errors.New("timeout waiting for slave transfer result")
|
||||
)
|
||||
@ -0,0 +1,121 @@
|
||||
package slaveinmaster
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
model "github.com/cloudreve/Cloudreve/v3/models"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/cluster"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/filesystem/driver"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/filesystem/fsctx"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/filesystem/response"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/mq"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/request"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/serializer"
|
||||
"io"
|
||||
"net/url"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Driver 影子存储策略，将上传任务指派给从机节点处理，并等待从机通知上传结果
|
||||
type Driver struct {
|
||||
node cluster.Node
|
||||
handler driver.Handler
|
||||
policy *model.Policy
|
||||
client request.Client
|
||||
}
|
||||
|
||||
// NewDriver 返回新的从机指派处理器
|
||||
func NewDriver(node cluster.Node, handler driver.Handler, policy *model.Policy) driver.Handler {
|
||||
var endpoint *url.URL
|
||||
if serverURL, err := url.Parse(node.DBModel().Server); err == nil {
|
||||
var controller *url.URL
|
||||
controller, _ = url.Parse("/api/v3/slave")
|
||||
endpoint = serverURL.ResolveReference(controller)
|
||||
}
|
||||
|
||||
signTTL := model.GetIntSetting("slave_api_timeout", 60)
|
||||
return &Driver{
|
||||
node: node,
|
||||
handler: handler,
|
||||
policy: policy,
|
||||
client: request.NewClient(
|
||||
request.WithMasterMeta(),
|
||||
request.WithTimeout(time.Duration(signTTL)*time.Second),
|
||||
request.WithCredential(node.SlaveAuthInstance(), int64(signTTL)),
|
||||
request.WithEndpoint(endpoint.String()),
|
||||
),
|
||||
}
|
||||
}
|
||||
|
||||
// Put 将ctx中指定的从机物理文件由从机上传到目标存储策略
|
||||
func (d *Driver) Put(ctx context.Context, file io.ReadCloser, dst string, size uint64) error {
|
||||
src, ok := ctx.Value(fsctx.SlaveSrcPath).(string)
|
||||
if !ok {
|
||||
return ErrSlaveSrcPathNotExist
|
||||
}
|
||||
|
||||
req := serializer.SlaveTransferReq{
|
||||
Src: src,
|
||||
Dst: dst,
|
||||
Policy: d.policy,
|
||||
}
|
||||
|
||||
body, err := json.Marshal(req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// 订阅转存结果
|
||||
resChan := mq.GlobalMQ.Subscribe(req.Hash(model.GetSettingByName("siteID")), 0)
|
||||
defer mq.GlobalMQ.Unsubscribe(req.Hash(model.GetSettingByName("siteID")), resChan)
|
||||
|
||||
res, err := d.client.Request("PUT", "task/transfer", bytes.NewReader(body)).
|
||||
CheckHTTPResponse(200).
|
||||
DecodeResponse()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if res.Code != 0 {
|
||||
return serializer.NewErrorFromResponse(res)
|
||||
}
|
||||
|
||||
// 等待转存结果或者超时
|
||||
waitTimeout := model.GetIntSetting("slave_transfer_timeout", 172800)
|
||||
select {
|
||||
case <-time.After(time.Duration(waitTimeout) * time.Second):
|
||||
return ErrWaitResultTimeout
|
||||
case msg := <-resChan:
|
||||
if msg.Event != serializer.SlaveTransferSuccess {
|
||||
return errors.New(msg.Content.(serializer.SlaveTransferResult).Error)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (d *Driver) Delete(ctx context.Context, files []string) ([]string, error) {
|
||||
return d.handler.Delete(ctx, files)
|
||||
}
|
||||
|
||||
func (d *Driver) Get(ctx context.Context, path string) (response.RSCloser, error) {
|
||||
return nil, ErrNotImplemented
|
||||
}
|
||||
|
||||
func (d *Driver) Thumb(ctx context.Context, path string) (*response.ContentResponse, error) {
|
||||
return nil, ErrNotImplemented
|
||||
}
|
||||
|
||||
func (d *Driver) Source(ctx context.Context, path string, url url.URL, ttl int64, isDownload bool, speed int) (string, error) {
|
||||
return "", ErrNotImplemented
|
||||
}
|
||||
|
||||
func (d *Driver) Token(ctx context.Context, ttl int64, callbackKey string) (serializer.UploadCredential, error) {
|
||||
return serializer.UploadCredential{}, ErrNotImplemented
|
||||
}
|
||||
|
||||
func (d *Driver) List(ctx context.Context, path string, recursive bool) ([]response.Object, error) {
|
||||
return nil, ErrNotImplemented
|
||||
}
|
||||
@ -0,0 +1,160 @@
|
||||
package mq
|
||||
|
||||
import (
|
||||
"encoding/gob"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/aria2/common"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/aria2/rpc"
|
||||
"strconv"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Message 消息事件正文
|
||||
type Message struct {
|
||||
// 消息触发者
|
||||
TriggeredBy string
|
||||
|
||||
// 事件标识
|
||||
Event string
|
||||
|
||||
// 消息正文
|
||||
Content interface{}
|
||||
}
|
||||
|
||||
type CallbackFunc func(Message)
|
||||
|
||||
// MQ 消息队列
|
||||
type MQ interface {
|
||||
rpc.Notifier
|
||||
|
||||
// 发布一个消息
|
||||
Publish(string, Message)
|
||||
|
||||
// 订阅一个消息主题
|
||||
Subscribe(string, int) <-chan Message
|
||||
|
||||
// 订阅一个消息主题，注册触发回调函数
|
||||
SubscribeCallback(string, CallbackFunc)
|
||||
|
||||
// 取消订阅一个消息主题
|
||||
Unsubscribe(string, <-chan Message)
|
||||
}
|
||||
|
||||
var GlobalMQ = NewMQ()
|
||||
|
||||
func NewMQ() MQ {
|
||||
return &inMemoryMQ{
|
||||
topics: make(map[string][]chan Message),
|
||||
callbacks: make(map[string][]CallbackFunc),
|
||||
}
|
||||
}
|
||||
|
||||
func init() {
|
||||
gob.Register(Message{})
|
||||
gob.Register([]rpc.Event{})
|
||||
}
|
||||
|
||||
type inMemoryMQ struct {
|
||||
topics map[string][]chan Message
|
||||
callbacks map[string][]CallbackFunc
|
||||
sync.RWMutex
|
||||
}
|
||||
|
||||
func (i *inMemoryMQ) Publish(topic string, message Message) {
|
||||
i.RLock()
|
||||
subscribersChan, okChan := i.topics[topic]
|
||||
subscribersCallback, okCallback := i.callbacks[topic]
|
||||
i.RUnlock()
|
||||
|
||||
if okChan {
|
||||
go func(subscribersChan []chan Message) {
|
||||
for i := 0; i < len(subscribersChan); i++ {
|
||||
select {
|
||||
case subscribersChan[i] <- message:
|
||||
case <-time.After(time.Millisecond * 500):
|
||||
}
|
||||
}
|
||||
}(subscribersChan)
|
||||
|
||||
}
|
||||
|
||||
if okCallback {
|
||||
for i := 0; i < len(subscribersCallback); i++ {
|
||||
go subscribersCallback[i](message)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (i *inMemoryMQ) Subscribe(topic string, buffer int) <-chan Message {
|
||||
ch := make(chan Message, buffer)
|
||||
i.Lock()
|
||||
i.topics[topic] = append(i.topics[topic], ch)
|
||||
i.Unlock()
|
||||
return ch
|
||||
}
|
||||
|
||||
func (i *inMemoryMQ) SubscribeCallback(topic string, callbackFunc CallbackFunc) {
|
||||
i.Lock()
|
||||
i.callbacks[topic] = append(i.callbacks[topic], callbackFunc)
|
||||
i.Unlock()
|
||||
}
|
||||
|
||||
func (i *inMemoryMQ) Unsubscribe(topic string, sub <-chan Message) {
|
||||
i.Lock()
|
||||
defer i.Unlock()
|
||||
|
||||
subscribers, ok := i.topics[topic]
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
|
||||
var newSubs []chan Message
|
||||
for _, subscriber := range subscribers {
|
||||
if subscriber == sub {
|
||||
continue
|
||||
}
|
||||
newSubs = append(newSubs, subscriber)
|
||||
}
|
||||
|
||||
i.topics[topic] = newSubs
|
||||
}
|
||||
|
||||
func (i *inMemoryMQ) Aria2Notify(events []rpc.Event, status int) {
|
||||
for _, event := range events {
|
||||
i.Publish(event.Gid, Message{
|
||||
TriggeredBy: event.Gid,
|
||||
Event: strconv.FormatInt(int64(status), 10),
|
||||
Content: events,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// OnDownloadStart 下载开始
|
||||
func (i *inMemoryMQ) OnDownloadStart(events []rpc.Event) {
|
||||
i.Aria2Notify(events, common.Downloading)
|
||||
}
|
||||
|
||||
// OnDownloadPause 下载暂停
|
||||
func (i *inMemoryMQ) OnDownloadPause(events []rpc.Event) {
|
||||
i.Aria2Notify(events, common.Paused)
|
||||
}
|
||||
|
||||
// OnDownloadStop 下载停止
|
||||
func (i *inMemoryMQ) OnDownloadStop(events []rpc.Event) {
|
||||
i.Aria2Notify(events, common.Canceled)
|
||||
}
|
||||
|
||||
// OnDownloadComplete 下载完成
|
||||
func (i *inMemoryMQ) OnDownloadComplete(events []rpc.Event) {
|
||||
i.Aria2Notify(events, common.Complete)
|
||||
}
|
||||
|
||||
// OnDownloadError 下载出错
|
||||
func (i *inMemoryMQ) OnDownloadError(events []rpc.Event) {
|
||||
i.Aria2Notify(events, common.Error)
|
||||
}
|
||||
|
||||
// OnBtDownloadComplete BT下载完成
|
||||
func (i *inMemoryMQ) OnBtDownloadComplete(events []rpc.Event) {
|
||||
i.Aria2Notify(events, common.Complete)
|
||||
}
|
||||
@ -0,0 +1,149 @@
|
||||
package mq
|
||||
|
||||
import (
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/aria2/rpc"
|
||||
"github.com/stretchr/testify/assert"
|
||||
"sync"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
func TestPublishAndSubscribe(t *testing.T) {
|
||||
t.Parallel()
|
||||
asserts := assert.New(t)
|
||||
mq := NewMQ()
|
||||
|
||||
// No subscriber
|
||||
{
|
||||
asserts.NotPanics(func() {
|
||||
mq.Publish("No subscriber", Message{})
|
||||
})
|
||||
}
|
||||
|
||||
// One channel subscriber
|
||||
{
|
||||
topic := "One channel subscriber"
|
||||
msg := Message{TriggeredBy: "Tester"}
|
||||
notifier := mq.Subscribe(topic, 0)
|
||||
mq.Publish(topic, msg)
|
||||
wg := sync.WaitGroup{}
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
wg.Done()
|
||||
msgRecv := <-notifier
|
||||
asserts.Equal(msg, msgRecv)
|
||||
}()
|
||||
wg.Wait()
|
||||
}
|
||||
|
||||
// two channel subscriber
|
||||
{
|
||||
topic := "two channel subscriber"
|
||||
msg := Message{TriggeredBy: "Tester"}
|
||||
notifier := mq.Subscribe(topic, 0)
|
||||
notifier2 := mq.Subscribe(topic, 0)
|
||||
mq.Publish(topic, msg)
|
||||
wg := sync.WaitGroup{}
|
||||
wg.Add(2)
|
||||
go func() {
|
||||
wg.Done()
|
||||
msgRecv := <-notifier
|
||||
asserts.Equal(msg, msgRecv)
|
||||
}()
|
||||
go func() {
|
||||
wg.Done()
|
||||
msgRecv := <-notifier2
|
||||
asserts.Equal(msg, msgRecv)
|
||||
}()
|
||||
wg.Wait()
|
||||
}
|
||||
|
||||
// two channel subscriber, one timeout
|
||||
{
|
||||
topic := "two channel subscriber, one timeout"
|
||||
msg := Message{TriggeredBy: "Tester"}
|
||||
mq.Subscribe(topic, 0)
|
||||
notifier2 := mq.Subscribe(topic, 0)
|
||||
mq.Publish(topic, msg)
|
||||
wg := sync.WaitGroup{}
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
wg.Done()
|
||||
msgRecv := <-notifier2
|
||||
asserts.Equal(msg, msgRecv)
|
||||
}()
|
||||
wg.Wait()
|
||||
}
|
||||
|
||||
// two channel subscriber, one unsubscribe
|
||||
{
|
||||
topic := "two channel subscriber, one unsubscribe"
|
||||
msg := Message{TriggeredBy: "Tester"}
|
||||
mq.Subscribe(topic, 0)
|
||||
notifier2 := mq.Subscribe(topic, 0)
|
||||
notifier := mq.Subscribe(topic, 0)
|
||||
mq.Unsubscribe(topic, notifier)
|
||||
mq.Publish(topic, msg)
|
||||
wg := sync.WaitGroup{}
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
wg.Done()
|
||||
msgRecv := <-notifier2
|
||||
asserts.Equal(msg, msgRecv)
|
||||
}()
|
||||
wg.Wait()
|
||||
|
||||
select {
|
||||
case <-notifier:
|
||||
t.Error()
|
||||
default:
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestAria2Interface(t *testing.T) {
|
||||
t.Parallel()
|
||||
asserts := assert.New(t)
|
||||
mq := NewMQ()
|
||||
var (
|
||||
OnDownloadStart int
|
||||
OnDownloadPause int
|
||||
OnDownloadStop int
|
||||
OnDownloadComplete int
|
||||
OnDownloadError int
|
||||
)
|
||||
l := sync.Mutex{}
|
||||
|
||||
mq.SubscribeCallback("TestAria2Interface", func(message Message) {
|
||||
asserts.Equal("TestAria2Interface", message.TriggeredBy)
|
||||
l.Lock()
|
||||
defer l.Unlock()
|
||||
switch message.Event {
|
||||
case "1":
|
||||
OnDownloadStart++
|
||||
case "2":
|
||||
OnDownloadPause++
|
||||
case "5":
|
||||
OnDownloadStop++
|
||||
case "4":
|
||||
OnDownloadComplete++
|
||||
case "3":
|
||||
OnDownloadError++
|
||||
}
|
||||
})
|
||||
|
||||
mq.OnDownloadStart([]rpc.Event{{"TestAria2Interface"}, {"TestAria2Interface"}})
|
||||
mq.OnDownloadPause([]rpc.Event{{"TestAria2Interface"}, {"TestAria2Interface"}})
|
||||
mq.OnDownloadStop([]rpc.Event{{"TestAria2Interface"}, {"TestAria2Interface"}})
|
||||
mq.OnDownloadComplete([]rpc.Event{{"TestAria2Interface"}, {"TestAria2Interface"}})
|
||||
mq.OnDownloadError([]rpc.Event{{"TestAria2Interface"}, {"TestAria2Interface"}})
|
||||
mq.OnBtDownloadComplete([]rpc.Event{{"TestAria2Interface"}, {"TestAria2Interface"}})
|
||||
|
||||
time.Sleep(time.Duration(500) * time.Millisecond)
|
||||
|
||||
asserts.Equal(2, OnDownloadStart)
|
||||
asserts.Equal(2, OnDownloadPause)
|
||||
asserts.Equal(2, OnDownloadStop)
|
||||
asserts.Equal(4, OnDownloadComplete)
|
||||
asserts.Equal(2, OnDownloadError)
|
||||
}
|
||||
@ -0,0 +1,110 @@
|
||||
package request
|
||||
|
||||
import (
|
||||
"context"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/auth"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"time"
|
||||
)
|
||||
|
||||
// Option 发送请求的额外设置
|
||||
type Option interface {
|
||||
apply(*options)
|
||||
}
|
||||
|
||||
type options struct {
|
||||
timeout time.Duration
|
||||
header http.Header
|
||||
sign auth.Auth
|
||||
signTTL int64
|
||||
ctx context.Context
|
||||
contentLength int64
|
||||
masterMeta bool
|
||||
endpoint *url.URL
|
||||
slaveNodeID string
|
||||
}
|
||||
|
||||
type optionFunc func(*options)
|
||||
|
||||
func (f optionFunc) apply(o *options) {
|
||||
f(o)
|
||||
}
|
||||
|
||||
func newDefaultOption() *options {
|
||||
return &options{
|
||||
header: http.Header{},
|
||||
timeout: time.Duration(30) * time.Second,
|
||||
contentLength: -1,
|
||||
}
|
||||
}
|
||||
|
||||
// WithTimeout 设置请求超时
|
||||
func WithTimeout(t time.Duration) Option {
|
||||
return optionFunc(func(o *options) {
|
||||
o.timeout = t
|
||||
})
|
||||
}
|
||||
|
||||
// WithContext 设置请求上下文
|
||||
func WithContext(c context.Context) Option {
|
||||
return optionFunc(func(o *options) {
|
||||
o.ctx = c
|
||||
})
|
||||
}
|
||||
|
||||
// WithCredential 对请求进行签名
|
||||
func WithCredential(instance auth.Auth, ttl int64) Option {
|
||||
return optionFunc(func(o *options) {
|
||||
o.sign = instance
|
||||
o.signTTL = ttl
|
||||
})
|
||||
}
|
||||
|
||||
// WithHeader 设置请求Header
|
||||
func WithHeader(header http.Header) Option {
|
||||
return optionFunc(func(o *options) {
|
||||
for k, v := range header {
|
||||
o.header[k] = v
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// WithoutHeader 设置清除请求Header
|
||||
func WithoutHeader(header []string) Option {
|
||||
return optionFunc(func(o *options) {
|
||||
for _, v := range header {
|
||||
delete(o.header, v)
|
||||
}
|
||||
|
||||
})
|
||||
}
|
||||
|
||||
// WithContentLength 设置请求大小
|
||||
func WithContentLength(s int64) Option {
|
||||
return optionFunc(func(o *options) {
|
||||
o.contentLength = s
|
||||
})
|
||||
}
|
||||
|
||||
// WithMasterMeta 请求时携带主机信息
|
||||
func WithMasterMeta() Option {
|
||||
return optionFunc(func(o *options) {
|
||||
o.masterMeta = true
|
||||
})
|
||||
}
|
||||
|
||||
// WithSlaveMeta 请求时携带从机信息
|
||||
func WithSlaveMeta(s string) Option {
|
||||
return optionFunc(func(o *options) {
|
||||
o.slaveNodeID = s
|
||||
})
|
||||
}
|
||||
|
||||
// Endpoint 使用同一的请求Endpoint
|
||||
func WithEndpoint(endpoint string) Option {
|
||||
endpointURL, _ := url.Parse(endpoint)
|
||||
return optionFunc(func(o *options) {
|
||||
o.endpoint = endpointURL
|
||||
})
|
||||
}
|
||||
@ -0,0 +1,35 @@
|
||||
package serializer
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/base64"
|
||||
"encoding/gob"
|
||||
)
|
||||
|
||||
// Response 基础序列化器
|
||||
type Response struct {
|
||||
Code int `json:"code"`
|
||||
Data interface{} `json:"data,omitempty"`
|
||||
Msg string `json:"msg"`
|
||||
Error string `json:"error,omitempty"`
|
||||
}
|
||||
|
||||
// NewResponseWithGobData 返回Data字段使用gob编码的Response
|
||||
func NewResponseWithGobData(data interface{}) Response {
|
||||
var w bytes.Buffer
|
||||
encoder := gob.NewEncoder(&w)
|
||||
if err := encoder.Encode(data); err != nil {
|
||||
return Err(CodeInternalSetting, "无法编码返回结果", err)
|
||||
}
|
||||
|
||||
return Response{Data: w.Bytes()}
|
||||
}
|
||||
|
||||
// GobDecode 将 Response 正文解码至目标指针
|
||||
func (r *Response) GobDecode(target interface{}) {
|
||||
src := r.Data.(string)
|
||||
raw := make([]byte, len(src)*len(src)/base64.StdEncoding.DecodedLen(len(src)))
|
||||
base64.StdEncoding.Decode(raw, []byte(src))
|
||||
decoder := gob.NewDecoder(bytes.NewBuffer(raw))
|
||||
decoder.Decode(target)
|
||||
}
|
||||
@ -0,0 +1,7 @@
|
||||
package slave
|
||||
|
||||
import "github.com/cloudreve/Cloudreve/v3/pkg/serializer"
|
||||
|
||||
var (
|
||||
ErrMasterNotFound = serializer.NewError(serializer.CodeMasterNotFound, "未知的主机节点", nil)
|
||||
)
|
||||
@ -0,0 +1,209 @@
|
||||
package slave
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/gob"
|
||||
"fmt"
|
||||
model "github.com/cloudreve/Cloudreve/v3/models"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/aria2/common"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/aria2/rpc"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/auth"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/cluster"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/mq"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/request"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/serializer"
|
||||
"github.com/jinzhu/gorm"
|
||||
"net/url"
|
||||
"sync"
|
||||
)
|
||||
|
||||
var DefaultController Controller
|
||||
|
||||
// Controller controls communications between master and slave
|
||||
type Controller interface {
|
||||
// Handle heartbeat sent from master
|
||||
HandleHeartBeat(*serializer.NodePingReq) (serializer.NodePingResp, error)
|
||||
|
||||
// Get Aria2 Instance by master node ID
|
||||
GetAria2Instance(string) (common.Aria2, error)
|
||||
|
||||
// Send event change message to master node
|
||||
SendNotification(string, string, mq.Message) error
|
||||
|
||||
// Submit async task into task pool
|
||||
SubmitTask(string, interface{}, string, func(interface{})) error
|
||||
|
||||
// Get master node info
|
||||
GetMasterInfo(string) (*MasterInfo, error)
|
||||
|
||||
// Get master OneDrive policy credential
|
||||
GetOneDriveToken(string, uint) (string, error)
|
||||
}
|
||||
|
||||
type slaveController struct {
|
||||
masters map[string]MasterInfo
|
||||
lock sync.RWMutex
|
||||
}
|
||||
|
||||
// info of master node
|
||||
type MasterInfo struct {
|
||||
ID string
|
||||
TTL int
|
||||
URL *url.URL
|
||||
// used to invoke aria2 rpc calls
|
||||
Instance cluster.Node
|
||||
Client request.Client
|
||||
|
||||
jobTracker map[string]bool
|
||||
}
|
||||
|
||||
func Init() {
|
||||
DefaultController = &slaveController{
|
||||
masters: make(map[string]MasterInfo),
|
||||
}
|
||||
gob.Register(rpc.StatusInfo{})
|
||||
}
|
||||
|
||||
func (c *slaveController) HandleHeartBeat(req *serializer.NodePingReq) (serializer.NodePingResp, error) {
|
||||
c.lock.Lock()
|
||||
defer c.lock.Unlock()
|
||||
|
||||
req.Node.AfterFind()
|
||||
|
||||
// close old node if exist
|
||||
origin, ok := c.masters[req.SiteID]
|
||||
|
||||
if (ok && req.IsUpdate) || !ok {
|
||||
if ok {
|
||||
origin.Instance.Kill()
|
||||
}
|
||||
|
||||
masterUrl, err := url.Parse(req.SiteURL)
|
||||
if err != nil {
|
||||
return serializer.NodePingResp{}, err
|
||||
}
|
||||
|
||||
c.masters[req.SiteID] = MasterInfo{
|
||||
ID: req.SiteID,
|
||||
URL: masterUrl,
|
||||
TTL: req.CredentialTTL,
|
||||
Client: request.NewClient(
|
||||
request.WithEndpoint(masterUrl.String()),
|
||||
request.WithSlaveMeta(fmt.Sprintf("%d", req.Node.ID)),
|
||||
request.WithCredential(auth.HMACAuth{
|
||||
SecretKey: []byte(req.Node.MasterKey),
|
||||
}, int64(req.CredentialTTL)),
|
||||
),
|
||||
jobTracker: make(map[string]bool),
|
||||
Instance: cluster.NewNodeFromDBModel(&model.Node{
|
||||
Model: gorm.Model{ID: req.Node.ID},
|
||||
MasterKey: req.Node.MasterKey,
|
||||
Type: model.MasterNodeType,
|
||||
Aria2Enabled: req.Node.Aria2Enabled,
|
||||
Aria2OptionsSerialized: req.Node.Aria2OptionsSerialized,
|
||||
}),
|
||||
}
|
||||
}
|
||||
|
||||
return serializer.NodePingResp{}, nil
|
||||
}
|
||||
|
||||
func (c *slaveController) GetAria2Instance(id string) (common.Aria2, error) {
|
||||
c.lock.RLock()
|
||||
defer c.lock.RUnlock()
|
||||
|
||||
if node, ok := c.masters[id]; ok {
|
||||
return node.Instance.GetAria2Instance(), nil
|
||||
}
|
||||
|
||||
return nil, ErrMasterNotFound
|
||||
}
|
||||
|
||||
func (c *slaveController) SendNotification(id, subject string, msg mq.Message) error {
|
||||
c.lock.RLock()
|
||||
|
||||
if node, ok := c.masters[id]; ok {
|
||||
c.lock.RUnlock()
|
||||
|
||||
body := bytes.Buffer{}
|
||||
enc := gob.NewEncoder(&body)
|
||||
if err := enc.Encode(&msg); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
res, err := node.Client.Request(
|
||||
"PUT",
|
||||
fmt.Sprintf("/api/v3/slave/notification/%s", subject),
|
||||
&body,
|
||||
).CheckHTTPResponse(200).DecodeResponse()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if res.Code != 0 {
|
||||
return serializer.NewErrorFromResponse(res)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
c.lock.RUnlock()
|
||||
return ErrMasterNotFound
|
||||
}
|
||||
|
||||
// SubmitTask 提交异步任务
|
||||
func (c *slaveController) SubmitTask(id string, job interface{}, hash string, submitter func(interface{})) error {
|
||||
c.lock.RLock()
|
||||
defer c.lock.RUnlock()
|
||||
|
||||
if node, ok := c.masters[id]; ok {
|
||||
if _, ok := node.jobTracker[hash]; ok {
|
||||
// 任务已存在，直接返回
|
||||
return nil
|
||||
}
|
||||
|
||||
submitter(job)
|
||||
return nil
|
||||
}
|
||||
|
||||
return ErrMasterNotFound
|
||||
}
|
||||
|
||||
// GetMasterInfo 获取主机节点信息
|
||||
func (c *slaveController) GetMasterInfo(id string) (*MasterInfo, error) {
|
||||
c.lock.RLock()
|
||||
defer c.lock.RUnlock()
|
||||
|
||||
if node, ok := c.masters[id]; ok {
|
||||
return &node, nil
|
||||
}
|
||||
|
||||
return nil, ErrMasterNotFound
|
||||
}
|
||||
|
||||
// GetOneDriveToken 获取主机OneDrive凭证
|
||||
func (c *slaveController) GetOneDriveToken(id string, policyID uint) (string, error) {
|
||||
c.lock.RLock()
|
||||
|
||||
if node, ok := c.masters[id]; ok {
|
||||
c.lock.RUnlock()
|
||||
|
||||
res, err := node.Client.Request(
|
||||
"GET",
|
||||
fmt.Sprintf("/api/v3/slave/credential/onedrive/%d", policyID),
|
||||
nil,
|
||||
).CheckHTTPResponse(200).DecodeResponse()
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
|
||||
if res.Code != 0 {
|
||||
return "", serializer.NewErrorFromResponse(res)
|
||||
}
|
||||
|
||||
return res.Data.(string), nil
|
||||
}
|
||||
|
||||
c.lock.RUnlock()
|
||||
return "", ErrMasterNotFound
|
||||
}
|
||||
@ -0,0 +1,145 @@
|
||||
package slavetask
|
||||
|
||||
import (
|
||||
"context"
|
||||
model "github.com/cloudreve/Cloudreve/v3/models"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/filesystem"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/filesystem/fsctx"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/mq"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/serializer"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/slave"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/task"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/util"
|
||||
"os"
|
||||
"path/filepath"
|
||||
)
|
||||
|
||||
// TransferTask 文件中转任务
|
||||
type TransferTask struct {
|
||||
Err *task.JobError
|
||||
Req *serializer.SlaveTransferReq
|
||||
MasterID string
|
||||
}
|
||||
|
||||
// Props 获取任务属性
|
||||
func (job *TransferTask) Props() string {
|
||||
return ""
|
||||
}
|
||||
|
||||
// Type 获取任务类型
|
||||
func (job *TransferTask) Type() int {
|
||||
return 0
|
||||
}
|
||||
|
||||
// Creator 获取创建者ID
|
||||
func (job *TransferTask) Creator() uint {
|
||||
return 0
|
||||
}
|
||||
|
||||
// Model 获取任务的数据库模型
|
||||
func (job *TransferTask) Model() *model.Task {
|
||||
return nil
|
||||
}
|
||||
|
||||
// SetStatus 设定状态
|
||||
func (job *TransferTask) SetStatus(status int) {
|
||||
}
|
||||
|
||||
// SetError 设定任务失败信息
|
||||
func (job *TransferTask) SetError(err *task.JobError) {
|
||||
job.Err = err
|
||||
|
||||
}
|
||||
|
||||
// SetErrorMsg 设定任务失败信息
|
||||
func (job *TransferTask) SetErrorMsg(msg string, err error) {
|
||||
jobErr := &task.JobError{Msg: msg}
|
||||
if err != nil {
|
||||
jobErr.Error = err.Error()
|
||||
}
|
||||
|
||||
job.SetError(jobErr)
|
||||
|
||||
notifyMsg := mq.Message{
|
||||
TriggeredBy: job.MasterID,
|
||||
Event: serializer.SlaveTransferFailed,
|
||||
Content: serializer.SlaveTransferResult{
|
||||
Error: err.Error(),
|
||||
},
|
||||
}
|
||||
|
||||
if err := slave.DefaultController.SendNotification(job.MasterID, job.Req.Hash(job.MasterID), notifyMsg); err != nil {
|
||||
util.Log().Warning("无法发送转存失败通知到从机, ", err)
|
||||
}
|
||||
}
|
||||
|
||||
// GetError 返回任务失败信息
|
||||
func (job *TransferTask) GetError() *task.JobError {
|
||||
return job.Err
|
||||
}
|
||||
|
||||
// Do 开始执行任务
|
||||
func (job *TransferTask) Do() {
|
||||
defer job.Recycle()
|
||||
|
||||
fs, err := filesystem.NewAnonymousFileSystem()
|
||||
if err != nil {
|
||||
job.SetErrorMsg("无法初始化匿名文件系统", err)
|
||||
return
|
||||
}
|
||||
|
||||
fs.Policy = job.Req.Policy
|
||||
if err := fs.DispatchHandler(); err != nil {
|
||||
job.SetErrorMsg("无法分发存储策略", err)
|
||||
return
|
||||
}
|
||||
|
||||
master, err := slave.DefaultController.GetMasterInfo(job.MasterID)
|
||||
if err != nil {
|
||||
job.SetErrorMsg("找不到主机节点", err)
|
||||
return
|
||||
}
|
||||
|
||||
fs.SwitchToShadowHandler(master.Instance, master.URL.String(), master.ID)
|
||||
ctx := context.WithValue(context.Background(), fsctx.DisableOverwrite, true)
|
||||
file, err := os.Open(util.RelativePath(job.Req.Src))
|
||||
if err != nil {
|
||||
job.SetErrorMsg("无法读取源文件", err)
|
||||
return
|
||||
}
|
||||
|
||||
defer file.Close()
|
||||
|
||||
// 获取源文件大小
|
||||
fi, err := file.Stat()
|
||||
if err != nil {
|
||||
job.SetErrorMsg("无法获取源文件大小", err)
|
||||
return
|
||||
}
|
||||
|
||||
size := fi.Size()
|
||||
|
||||
err = fs.Handler.Put(ctx, file, job.Req.Dst, uint64(size))
|
||||
if err != nil {
|
||||
job.SetErrorMsg("文件上传失败", err)
|
||||
return
|
||||
}
|
||||
|
||||
msg := mq.Message{
|
||||
TriggeredBy: job.MasterID,
|
||||
Event: serializer.SlaveTransferSuccess,
|
||||
Content: serializer.SlaveTransferResult{},
|
||||
}
|
||||
|
||||
if err := slave.DefaultController.SendNotification(job.MasterID, job.Req.Hash(job.MasterID), msg); err != nil {
|
||||
util.Log().Warning("无法发送转存成功通知到从机, ", err)
|
||||
}
|
||||
}
|
||||
|
||||
// Recycle 回收临时文件
|
||||
func (job *TransferTask) Recycle() {
|
||||
err := os.RemoveAll(filepath.Dir(job.Req.Src))
|
||||
if err != nil {
|
||||
util.Log().Warning("无法删除中转临时目录[%s], %s", job.Req.Src, err)
|
||||
}
|
||||
}
|
||||
@ -1,43 +1,71 @@
|
||||
package admin
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"encoding/json"
|
||||
model "github.com/cloudreve/Cloudreve/v3/models"
|
||||
"net/url"
|
||||
"time"
|
||||
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/aria2"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/auth"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/request"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/serializer"
|
||||
)
|
||||
|
||||
// Aria2TestService aria2连接测试服务
|
||||
type Aria2TestService struct {
|
||||
Server string `json:"server" binding:"required"`
|
||||
Token string `json:"token"`
|
||||
Server string `json:"server" binding:"required"`
|
||||
RPC string `json:"rpc" binding:"required"`
|
||||
Secret string `json:"secret" binding:"required"`
|
||||
Token string `json:"token"`
|
||||
Type model.ModelType `json:"type"`
|
||||
}
|
||||
|
||||
// Test 测试aria2连接
|
||||
func (service *Aria2TestService) Test() serializer.Response {
|
||||
testRPC := aria2.RPCService{}
|
||||
|
||||
// 解析RPC服务地址
|
||||
server, err := url.Parse(service.Server)
|
||||
func (service *Aria2TestService) TestMaster() serializer.Response {
|
||||
res, err := aria2.TestRPCConnection(service.RPC, service.Token, 5)
|
||||
if err != nil {
|
||||
return serializer.ParamErr("无法解析 aria2 RPC 服务地址, "+err.Error(), nil)
|
||||
return serializer.ParamErr(err.Error(), err)
|
||||
}
|
||||
server.Path = "/jsonrpc"
|
||||
|
||||
if err := testRPC.Init(server.String(), service.Token, 5, map[string]interface{}{}); err != nil {
|
||||
return serializer.ParamErr("无法初始化连接, "+err.Error(), nil)
|
||||
if res.Version == "" {
|
||||
return serializer.ParamErr("RPC 服务返回非预期响应", nil)
|
||||
}
|
||||
|
||||
defer testRPC.Caller.Close()
|
||||
return serializer.Response{Data: res.Version}
|
||||
}
|
||||
|
||||
info, err := testRPC.Caller.GetVersion()
|
||||
func (service *Aria2TestService) TestSlave() serializer.Response {
|
||||
slave, err := url.Parse(service.Server)
|
||||
if err != nil {
|
||||
return serializer.ParamErr("无法请求 RPC 服务, "+err.Error(), nil)
|
||||
return serializer.ParamErr("无法解析从机端地址，"+err.Error(), nil)
|
||||
}
|
||||
|
||||
if info.Version == "" {
|
||||
return serializer.ParamErr("RPC 服务返回非预期响应", nil)
|
||||
controller, _ := url.Parse("/api/v3/slave/ping/aria2")
|
||||
|
||||
// 请求正文
|
||||
service.Type = model.MasterNodeType
|
||||
bodyByte, _ := json.Marshal(service)
|
||||
|
||||
r := request.NewClient()
|
||||
res, err := r.Request(
|
||||
"POST",
|
||||
slave.ResolveReference(controller).String(),
|
||||
bytes.NewReader(bodyByte),
|
||||
request.WithTimeout(time.Duration(10)*time.Second),
|
||||
request.WithCredential(
|
||||
auth.HMACAuth{SecretKey: []byte(service.Secret)},
|
||||
int64(model.GetIntSetting("slave_api_timeout", 60)),
|
||||
),
|
||||
).DecodeResponse()
|
||||
if err != nil {
|
||||
return serializer.ParamErr("无连接到从机，"+err.Error(), nil)
|
||||
}
|
||||
|
||||
if res.Code != 0 {
|
||||
return serializer.ParamErr("成功接到从机，但是从机返回："+res.Msg, nil)
|
||||
}
|
||||
|
||||
return serializer.Response{Data: info.Version}
|
||||
return serializer.Response{Data: res.Data.(string)}
|
||||
}
|
||||
|
||||
@ -0,0 +1,138 @@
|
||||
package admin
|
||||
|
||||
import (
|
||||
model "github.com/cloudreve/Cloudreve/v3/models"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/cluster"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/serializer"
|
||||
"strings"
|
||||
)
|
||||
|
||||
// AddNodeService 节点添加服务
|
||||
type AddNodeService struct {
|
||||
Node model.Node `json:"node" binding:"required"`
|
||||
}
|
||||
|
||||
// Add 添加节点
|
||||
func (service *AddNodeService) Add() serializer.Response {
|
||||
if service.Node.ID > 0 {
|
||||
if err := model.DB.Save(&service.Node).Error; err != nil {
|
||||
return serializer.ParamErr("节点保存失败", err)
|
||||
}
|
||||
} else {
|
||||
if err := model.DB.Create(&service.Node).Error; err != nil {
|
||||
return serializer.ParamErr("节点添加失败", err)
|
||||
}
|
||||
}
|
||||
|
||||
return serializer.Response{Data: service.Node.ID}
|
||||
}
|
||||
|
||||
// Nodes 列出从机节点
|
||||
func (service *AdminListService) Nodes() serializer.Response {
|
||||
var res []model.Node
|
||||
total := 0
|
||||
|
||||
tx := model.DB.Model(&model.Node{})
|
||||
if service.OrderBy != "" {
|
||||
tx = tx.Order(service.OrderBy)
|
||||
}
|
||||
|
||||
for k, v := range service.Conditions {
|
||||
tx = tx.Where(k+" = ?", v)
|
||||
}
|
||||
|
||||
if len(service.Searches) > 0 {
|
||||
search := ""
|
||||
for k, v := range service.Searches {
|
||||
search += k + " like '%" + v + "%' OR "
|
||||
}
|
||||
search = strings.TrimSuffix(search, " OR ")
|
||||
tx = tx.Where(search)
|
||||
}
|
||||
|
||||
// 计算总数用于分页
|
||||
tx.Count(&total)
|
||||
|
||||
// 查询记录
|
||||
tx.Limit(service.PageSize).Offset((service.Page - 1) * service.PageSize).Find(&res)
|
||||
|
||||
isActive := make(map[uint]bool)
|
||||
for i := 0; i < len(res); i++ {
|
||||
if node := cluster.Default.GetNodeByID(res[i].ID); node != nil {
|
||||
isActive[res[i].ID] = node.IsActive()
|
||||
}
|
||||
}
|
||||
|
||||
return serializer.Response{Data: map[string]interface{}{
|
||||
"total": total,
|
||||
"items": res,
|
||||
"active": isActive,
|
||||
}}
|
||||
}
|
||||
|
||||
// ToggleNodeService 开关节点服务
|
||||
type ToggleNodeService struct {
|
||||
ID uint `uri:"id"`
|
||||
Desired model.NodeStatus `uri:"desired"`
|
||||
}
|
||||
|
||||
// Toggle 开关节点
|
||||
func (service *ToggleNodeService) Toggle() serializer.Response {
|
||||
node, err := model.GetNodeByID(service.ID)
|
||||
if err != nil {
|
||||
return serializer.DBErr("找不到节点", err)
|
||||
}
|
||||
|
||||
// 是否为系统节点
|
||||
if node.ID <= 1 {
|
||||
return serializer.Err(serializer.CodeNoPermissionErr, "系统节点无法更改", err)
|
||||
}
|
||||
|
||||
if err = node.SetStatus(service.Desired); err != nil {
|
||||
return serializer.DBErr("无法更改节点状态", err)
|
||||
}
|
||||
|
||||
if service.Desired == model.NodeActive {
|
||||
cluster.Default.Add(&node)
|
||||
} else {
|
||||
cluster.Default.Delete(node.ID)
|
||||
}
|
||||
|
||||
return serializer.Response{}
|
||||
}
|
||||
|
||||
// NodeService 节点ID服务
|
||||
type NodeService struct {
|
||||
ID uint `uri:"id" json:"id" binding:"required"`
|
||||
}
|
||||
|
||||
// Delete 删除节点
|
||||
func (service *NodeService) Delete() serializer.Response {
|
||||
// 查找用户组
|
||||
node, err := model.GetNodeByID(service.ID)
|
||||
if err != nil {
|
||||
return serializer.Err(serializer.CodeNotFound, "节点不存在", err)
|
||||
}
|
||||
|
||||
// 是否为系统节点
|
||||
if node.ID <= 1 {
|
||||
return serializer.Err(serializer.CodeNoPermissionErr, "系统节点无法删除", err)
|
||||
}
|
||||
|
||||
cluster.Default.Delete(node.ID)
|
||||
if err := model.DB.Delete(&node).Error; err != nil {
|
||||
return serializer.DBErr("无法删除节点", err)
|
||||
}
|
||||
|
||||
return serializer.Response{}
|
||||
}
|
||||
|
||||
// Get 获取节点详情
|
||||
func (service *NodeService) Get() serializer.Response {
|
||||
node, err := model.GetNodeByID(service.ID)
|
||||
if err != nil {
|
||||
return serializer.Err(serializer.CodeNotFound, "节点不存在", err)
|
||||
}
|
||||
|
||||
return serializer.Response{Data: node}
|
||||
}
|
||||
@ -0,0 +1,166 @@
|
||||
package explorer
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/base64"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
model "github.com/cloudreve/Cloudreve/v3/models"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/filesystem"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/filesystem/fsctx"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/serializer"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/slave"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/task"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/task/slavetask"
|
||||
"github.com/gin-gonic/gin"
|
||||
"github.com/jinzhu/gorm"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"time"
|
||||
)
|
||||
|
||||
// SlaveDownloadService 从机文件下載服务
|
||||
type SlaveDownloadService struct {
|
||||
PathEncoded string `uri:"path" binding:"required"`
|
||||
Name string `uri:"name" binding:"required"`
|
||||
Speed int `uri:"speed" binding:"min=0"`
|
||||
}
|
||||
|
||||
// SlaveFileService 从机单文件文件相关服务
|
||||
type SlaveFileService struct {
|
||||
PathEncoded string `uri:"path" binding:"required"`
|
||||
}
|
||||
|
||||
// SlaveFilesService 从机多文件相关服务
|
||||
type SlaveFilesService struct {
|
||||
Files []string `json:"files" binding:"required,gt=0"`
|
||||
}
|
||||
|
||||
// SlaveListService 从机列表服务
|
||||
type SlaveListService struct {
|
||||
Path string `json:"path" binding:"required,min=1,max=65535"`
|
||||
Recursive bool `json:"recursive"`
|
||||
}
|
||||
|
||||
// ServeFile 通过签名的URL下载从机文件
|
||||
func (service *SlaveDownloadService) ServeFile(ctx context.Context, c *gin.Context, isDownload bool) serializer.Response {
|
||||
// 创建文件系统
|
||||
fs, err := filesystem.NewAnonymousFileSystem()
|
||||
if err != nil {
|
||||
return serializer.Err(serializer.CodePolicyNotAllowed, err.Error(), err)
|
||||
}
|
||||
defer fs.Recycle()
|
||||
|
||||
// 解码文件路径
|
||||
fileSource, err := base64.RawURLEncoding.DecodeString(service.PathEncoded)
|
||||
if err != nil {
|
||||
return serializer.ParamErr("无法解析的文件地址", err)
|
||||
}
|
||||
|
||||
// 根据URL里的信息创建一个文件对象和用户对象
|
||||
file := model.File{
|
||||
Name: service.Name,
|
||||
SourceName: string(fileSource),
|
||||
Policy: model.Policy{
|
||||
Model: gorm.Model{ID: 1},
|
||||
Type: "local",
|
||||
},
|
||||
}
|
||||
fs.User = &model.User{
|
||||
Group: model.Group{SpeedLimit: service.Speed},
|
||||
}
|
||||
fs.FileTarget = []model.File{file}
|
||||
|
||||
// 开始处理下载
|
||||
ctx = context.WithValue(ctx, fsctx.GinCtx, c)
|
||||
rs, err := fs.GetDownloadContent(ctx, 0)
|
||||
if err != nil {
|
||||
return serializer.Err(serializer.CodeNotSet, err.Error(), err)
|
||||
}
|
||||
defer rs.Close()
|
||||
|
||||
// 设置下载文件名
|
||||
if isDownload {
|
||||
c.Header("Content-Disposition", "attachment; filename=\""+url.PathEscape(fs.FileTarget[0].Name)+"\"")
|
||||
}
|
||||
|
||||
// 发送文件
|
||||
http.ServeContent(c.Writer, c.Request, fs.FileTarget[0].Name, time.Now(), rs)
|
||||
|
||||
return serializer.Response{
|
||||
Code: 0,
|
||||
}
|
||||
}
|
||||
|
||||
// Delete 通过签名的URL删除从机文件
|
||||
func (service *SlaveFilesService) Delete(ctx context.Context, c *gin.Context) serializer.Response {
|
||||
// 创建文件系统
|
||||
fs, err := filesystem.NewAnonymousFileSystem()
|
||||
if err != nil {
|
||||
return serializer.Err(serializer.CodePolicyNotAllowed, err.Error(), err)
|
||||
}
|
||||
defer fs.Recycle()
|
||||
|
||||
// 删除文件
|
||||
failed, err := fs.Handler.Delete(ctx, service.Files)
|
||||
if err != nil {
|
||||
// 将Data字段写为字符串方便主控端解析
|
||||
data, _ := json.Marshal(serializer.RemoteDeleteRequest{Files: failed})
|
||||
|
||||
return serializer.Response{
|
||||
Code: serializer.CodeNotFullySuccess,
|
||||
Data: string(data),
|
||||
Msg: fmt.Sprintf("有 %d 个文件未能成功删除", len(failed)),
|
||||
Error: err.Error(),
|
||||
}
|
||||
}
|
||||
return serializer.Response{Code: 0}
|
||||
}
|
||||
|
||||
// Thumb 通过签名URL获取从机文件缩略图
|
||||
func (service *SlaveFileService) Thumb(ctx context.Context, c *gin.Context) serializer.Response {
|
||||
// 创建文件系统
|
||||
fs, err := filesystem.NewAnonymousFileSystem()
|
||||
if err != nil {
|
||||
return serializer.Err(serializer.CodePolicyNotAllowed, err.Error(), err)
|
||||
}
|
||||
defer fs.Recycle()
|
||||
|
||||
// 解码文件路径
|
||||
fileSource, err := base64.RawURLEncoding.DecodeString(service.PathEncoded)
|
||||
if err != nil {
|
||||
return serializer.ParamErr("无法解析的文件地址", err)
|
||||
}
|
||||
fs.FileTarget = []model.File{{SourceName: string(fileSource), PicInfo: "1,1"}}
|
||||
|
||||
// 获取缩略图
|
||||
resp, err := fs.GetThumb(ctx, 0)
|
||||
if err != nil {
|
||||
return serializer.Err(serializer.CodeNotSet, "无法获取缩略图", err)
|
||||
}
|
||||
|
||||
defer resp.Content.Close()
|
||||
http.ServeContent(c.Writer, c.Request, "thumb.png", time.Now(), resp.Content)
|
||||
|
||||
return serializer.Response{Code: 0}
|
||||
}
|
||||
|
||||
// CreateTransferTask 创建从机文件转存任务
|
||||
func CreateTransferTask(c *gin.Context, req *serializer.SlaveTransferReq) serializer.Response {
|
||||
if id, ok := c.Get("MasterSiteID"); ok {
|
||||
job := &slavetask.TransferTask{
|
||||
Req: req,
|
||||
MasterID: id.(string),
|
||||
}
|
||||
|
||||
if err := slave.DefaultController.SubmitTask(job.MasterID, job, req.Hash(job.MasterID), func(job interface{}) {
|
||||
task.TaskPoll.Submit(job.(task.Job))
|
||||
}); err != nil {
|
||||
return serializer.Err(serializer.CodeInternalSetting, "任务创建失败", err)
|
||||
}
|
||||
|
||||
return serializer.Response{}
|
||||
}
|
||||
|
||||
return serializer.ParamErr("未知的主机节点ID", nil)
|
||||
}
|
||||
@ -0,0 +1,62 @@
|
||||
package node
|
||||
|
||||
import (
|
||||
"encoding/gob"
|
||||
model "github.com/cloudreve/Cloudreve/v3/models"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/filesystem/driver/onedrive"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/mq"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/serializer"
|
||||
"github.com/cloudreve/Cloudreve/v3/pkg/slave"
|
||||
"github.com/gin-gonic/gin"
|
||||
)
|
||||
|
||||
type SlaveNotificationService struct {
|
||||
Subject string `uri:"subject" binding:"required"`
|
||||
}
|
||||
|
||||
type OneDriveCredentialService struct {
|
||||
PolicyID uint `uri:"id" binding:"required"`
|
||||
}
|
||||
|
||||
func HandleMasterHeartbeat(req *serializer.NodePingReq) serializer.Response {
|
||||
res, err := slave.DefaultController.HandleHeartBeat(req)
|
||||
if err != nil {
|
||||
return serializer.Err(serializer.CodeInternalSetting, "Cannot initialize slave controller", err)
|
||||
}
|
||||
|
||||
return serializer.Response{
|
||||
Code: 0,
|
||||
Data: res,
|
||||
}
|
||||
}
|
||||
|
||||
// HandleSlaveNotificationPush 转发从机的消息通知到本机消息队列
|
||||
func (s *SlaveNotificationService) HandleSlaveNotificationPush(c *gin.Context) serializer.Response {
|
||||
var msg mq.Message
|
||||
dec := gob.NewDecoder(c.Request.Body)
|
||||
if err := dec.Decode(&msg); err != nil {
|
||||
return serializer.ParamErr("Cannot parse notification message", err)
|
||||
}
|
||||
|
||||
mq.GlobalMQ.Publish(s.Subject, msg)
|
||||
return serializer.Response{}
|
||||
}
|
||||
|
||||
// Get 获取主机OneDrive策略的AccessToken
|
||||
func (s *OneDriveCredentialService) Get(c *gin.Context) serializer.Response {
|
||||
policy, err := model.GetPolicyByID(s.PolicyID)
|
||||
if err != nil {
|
||||
return serializer.Err(serializer.CodeNotFound, "Cannot found storage policy", err)
|
||||
}
|
||||
|
||||
client, err := onedrive.NewClient(&policy)
|
||||
if err != nil {
|
||||
return serializer.Err(serializer.CodeInternalSetting, "Cannot initialize OneDrive client", err)
|
||||
}
|
||||
|
||||
if err := client.UpdateCredential(c); err != nil {
|
||||
return serializer.Err(serializer.CodeInternalSetting, "Cannot refresh OneDrive credential", err)
|
||||
}
|
||||
|
||||
return serializer.Response{Data: client.Credential.AccessToken}
|
||||
}
|
||||
Loading…
Reference in new issue