Compare commits

..

28 Commits

Author SHA1 Message Date
刘祥超
5d00d53ea5 增加ACME错误信息字段长度 2021-10-30 15:47:59 +08:00
刘祥超
dcf04d64bd 更新SQL 2021-10-29 14:06:12 +08:00
刘祥超
d36f7d01df 增加套餐相关代码 2021-10-29 14:02:40 +08:00
刘祥超
a0ff24adb6 增加在IP名单中使用ipFrom和ipTo查找IP的API 2021-10-26 09:17:33 +08:00
刘祥超
5597a0af6c 创建WAF分组时也记录代号 2021-10-25 19:02:20 +08:00
刘祥超
c196a85a59 增加为WAF分组添加规则集的API 2021-10-25 12:01:16 +08:00
刘祥超
ba638d4e1d 优化代码 2021-10-22 13:40:04 +08:00
刘祥超
7f8abccd2a 优化代码 2021-10-22 13:38:18 +08:00
刘祥超
c0f540cc2c 更新SQL 2021-10-22 12:40:30 +08:00
刘祥超
bd905ff1a9 可以在IP名单中搜索IP 2021-10-22 12:18:53 +08:00
刘祥超
3b8d1b4cd8 实现单个服务的带宽限制(商业版) 2021-10-21 17:10:53 +08:00
刘祥超
f86180b93c 将HTTP Header中Edge-改成X-Edge- 2021-10-19 19:49:06 +08:00
刘祥超
cfb1864fd2 健康检查支持UserAgent和是否基础请求设置 2021-10-19 16:31:05 +08:00
刘祥超
1d0a66a156 上传SQL 2021-10-18 09:09:32 +08:00
刘祥超
b4d4f6460e 增加PURGE某个URL缓存功能 2021-10-17 20:22:14 +08:00
刘祥超
e2de9799c0 增加清除服务缓存API 2021-10-17 17:12:30 +08:00
刘祥超
9b9c6471f7 增加支持服务CNAME选项/提供重新生成服务CNAME API 2021-10-16 12:02:42 +08:00
刘祥超
d30b10baee 只讲错误级别的节点运行日志设置为未读 2021-10-16 10:26:47 +08:00
刘祥超
55f7189a1c 运行日志显示未读的日志数量 2021-10-15 12:54:31 +08:00
刘祥超
0ff6fb002d 域名小时统计只保留7天 2021-10-15 10:16:14 +08:00
刘祥超
a5d34565c5 节点日志增加是否已读标记 2021-10-14 17:29:54 +08:00
刘祥超
3c17ba0a8b 修复华为云DNS TXT记录值不加引号无法添加的问题 2021-10-13 17:53:14 +08:00
刘祥超
a8c0c64071 更新SQL 2021-10-13 10:12:10 +08:00
刘祥超
82ed22a464 支持PROXY Protocol 2021-10-12 20:18:35 +08:00
刘祥超
b190479d44 集群API中增加时区信息 2021-10-12 14:40:24 +08:00
刘祥超
ed7b586137 可以在集群中指定节点时区 2021-10-12 11:44:24 +08:00
刘祥超
ea19635fe5 修复同属多集群下的节点无法删除线路的Bug 2021-10-12 08:40:39 +08:00
刘祥超
29cd7da6e4 版本号改为0.3.3 2021-10-12 08:38:00 +08:00
46 changed files with 1546 additions and 173 deletions

View File

@@ -1,3 +1,5 @@
// Copyright 2021 Liuxiangchao iwind.liu@gmail.com. All rights reserved.
package accesslogs
import (
@@ -5,7 +7,6 @@ import (
"github.com/TeaOSLab/EdgeAPI/internal/db/models"
"github.com/TeaOSLab/EdgeAPI/internal/errors"
"github.com/TeaOSLab/EdgeAPI/internal/remotelogs"
"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs"
"github.com/iwind/TeaGo/Tea"
"github.com/iwind/TeaGo/lists"
@@ -49,23 +50,6 @@ func (this *StorageManager) Start() {
}
}
// 写入日志
func (this *StorageManager) Write(policyId int64, accessLogs []*pb.HTTPAccessLog) error {
this.locker.Lock()
storage, ok := this.storageMap[policyId]
this.locker.Unlock()
if !ok {
return nil
}
if !storage.IsOk() {
return nil
}
return storage.Write(accessLogs)
}
// Loop 更新
func (this *StorageManager) Loop() error {
policies, err := models.SharedHTTPAccessLogPolicyDAO.FindAllEnabledAndOnPolicies(nil)

View File

@@ -0,0 +1,15 @@
// Copyright 2021 Liuxiangchao iwind.liu@gmail.com. All rights reserved.
//go:build community
// +build community
package accesslogs
import (
"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
)
// 写入日志
func (this *StorageManager) Write(policyId int64, accessLogs []*pb.HTTPAccessLog) error {
return nil
}

View File

@@ -1,7 +1,7 @@
package teaconst
const (
Version = "0.3.2"
Version = "0.3.3"
ProductName = "Edge API"
ProcessName = "edge-api"
@@ -18,7 +18,7 @@ const (
// 其他节点版本号,用来检测是否有需要升级的节点
NodeVersion = "0.3.2"
NodeVersion = "0.3.3"
UserNodeVersion = "0.0.10"
AuthorityNodeVersion = "0.0.2"
MonitorNodeVersion = "0.0.3"

View File

@@ -93,16 +93,3 @@ func (this *AuthorityKeyDAO) ResetKey(tx *dbs.Tx) error {
Delete()
return err
}
// IsPlus 判断是否为企业版
func (this *AuthorityKeyDAO) IsPlus(tx *dbs.Tx) (bool, error) {
key, err := this.ReadKey(tx)
if err != nil {
return false, err
}
if key == nil {
return false, nil
}
teaconst.IsPlus = key.DayTo >= timeutil.Format("Y-m-d")
return teaconst.IsPlus, nil
}

View File

@@ -0,0 +1,14 @@
// Copyright 2021 Liuxiangchao iwind.liu@gmail.com. All rights reserved.
//go:build community
// +build community
package authority
import (
"github.com/iwind/TeaGo/dbs"
)
// IsPlus 判断是否为企业版
func (this *AuthorityKeyDAO) IsPlus(tx *dbs.Tx) (bool, error) {
return false, nil
}

View File

@@ -163,11 +163,12 @@ func (this *HTTPFirewallRuleGroupDAO) UpdateGroupIsOn(tx *dbs.Tx, groupId int64,
}
// CreateGroup 创建分组
func (this *HTTPFirewallRuleGroupDAO) CreateGroup(tx *dbs.Tx, isOn bool, name string, description string) (int64, error) {
func (this *HTTPFirewallRuleGroupDAO) CreateGroup(tx *dbs.Tx, isOn bool, name string, code string, description string) (int64, error) {
op := NewHTTPFirewallRuleGroupOperator()
op.State = HTTPFirewallRuleStateEnabled
op.IsOn = isOn
op.Name = name
op.Code = code
op.Description = description
err := this.Save(tx, op)
if err != nil {
@@ -194,13 +195,13 @@ func (this *HTTPFirewallRuleGroupDAO) UpdateGroup(tx *dbs.Tx, groupId int64, isO
}
// UpdateGroupSets 修改分组中的规则集
func (this *HTTPFirewallRuleGroupDAO) UpdateGroupSets(tx *dbs.Tx, groupId int64, setsJSON []byte) error {
func (this *HTTPFirewallRuleGroupDAO) UpdateGroupSets(tx *dbs.Tx, groupId int64, setRefsJSON []byte) error {
if groupId <= 0 {
return errors.New("invalid groupId")
}
op := NewHTTPFirewallRuleGroupOperator()
op.Id = groupId
op.Sets = setsJSON
op.Sets = setRefsJSON
err := this.Save(tx, op)
if err != nil {
return err

View File

@@ -177,18 +177,39 @@ func (this *IPItemDAO) UpdateIPItem(tx *dbs.Tx, itemId int64, ipFrom string, ipT
}
// CountIPItemsWithListId 计算IP数量
func (this *IPItemDAO) CountIPItemsWithListId(tx *dbs.Tx, listId int64) (int64, error) {
return this.Query(tx).
func (this *IPItemDAO) CountIPItemsWithListId(tx *dbs.Tx, listId int64, ipFrom string, ipTo string, keyword string) (int64, error) {
var query = this.Query(tx).
State(IPItemStateEnabled).
Attr("listId", listId).
Count()
Attr("listId", listId)
if len(keyword) > 0 {
query.Where("(ipFrom LIKE :keyword OR ipTo LIKE :keyword)").
Param("keyword", "%"+keyword+"%")
}
if len(ipFrom) > 0 {
query.Attr("ipFrom", ipFrom)
}
if len(ipTo) > 0 {
query.Attr("ipTo", ipTo)
}
return query.Count()
}
// ListIPItemsWithListId 查找IP列表
func (this *IPItemDAO) ListIPItemsWithListId(tx *dbs.Tx, listId int64, offset int64, size int64) (result []*IPItem, err error) {
_, err = this.Query(tx).
func (this *IPItemDAO) ListIPItemsWithListId(tx *dbs.Tx, listId int64, keyword string, ipFrom string, ipTo string, offset int64, size int64) (result []*IPItem, err error) {
var query = this.Query(tx).
State(IPItemStateEnabled).
Attr("listId", listId).
Attr("listId", listId)
if len(keyword) > 0 {
query.Where("(ipFrom LIKE :keyword OR ipTo LIKE :keyword)").
Param("keyword", "%"+keyword+"%")
}
if len(ipFrom) > 0 {
query.Attr("ipFrom", ipFrom)
}
if len(ipTo) > 0 {
query.Attr("ipTo", ipTo)
}
_, err = query.
DescPk().
Slice(&result).
Offset(offset).

View File

@@ -1,6 +1,7 @@
package models
import (
teaconst "github.com/TeaOSLab/EdgeAPI/internal/const"
"github.com/TeaOSLab/EdgeAPI/internal/errors"
"github.com/TeaOSLab/EdgeAPI/internal/remotelogs"
"github.com/TeaOSLab/EdgeCommon/pkg/nodeconfigs"
@@ -94,6 +95,10 @@ func (this *MessageTaskDAO) FindEnabledMessageTask(tx *dbs.Tx, id int64) (*Messa
// CreateMessageTask 创建任务
func (this *MessageTaskDAO) CreateMessageTask(tx *dbs.Tx, recipientId int64, instanceId int64, user string, subject string, body string, isPrimary bool) (int64, error) {
if !teaconst.IsPlus {
return 0, nil
}
var hash = stringutil.Md5(types.String(recipientId) + "@" + types.String(instanceId) + "@" + user + "@" + subject + "@" + types.String(isPrimary))
recipientInstanceId, err := SharedMessageRecipientDAO.FindRecipientInstanceId(tx, recipientId)
if err != nil {
@@ -196,6 +201,10 @@ func (this *MessageTaskDAO) UpdateMessageTaskStatus(tx *dbs.Tx, taskId int64, st
// CreateMessageTasks 从集群、节点或者服务中创建任务
func (this *MessageTaskDAO) CreateMessageTasks(tx *dbs.Tx, role nodeconfigs.NodeRole, clusterId int64, nodeId int64, serverId int64, messageType MessageType, subject string, body string) error {
if !teaconst.IsPlus {
return nil
}
receivers, err := SharedMessageReceiverDAO.FindEnabledBestFitReceivers(tx, role, clusterId, nodeId, serverId, messageType)
if err != nil {
return err

View File

@@ -177,7 +177,7 @@ func (this *NodeClusterDAO) CreateCluster(tx *dbs.Tx, adminId int64, name string
}
// UpdateCluster 修改集群
func (this *NodeClusterDAO) UpdateCluster(tx *dbs.Tx, clusterId int64, name string, grantId int64, installDir string) error {
func (this *NodeClusterDAO) UpdateCluster(tx *dbs.Tx, clusterId int64, name string, grantId int64, installDir string, timezone string) error {
if clusterId <= 0 {
return errors.New("invalid clusterId")
}
@@ -186,8 +186,12 @@ func (this *NodeClusterDAO) UpdateCluster(tx *dbs.Tx, clusterId int64, name stri
op.Name = name
op.GrantId = grantId
op.InstallDir = installDir
op.TimeZone = timezone
err := this.Save(tx, op)
return err
if err != nil {
return err
}
return this.NotifyUpdate(tx, clusterId)
}
// CountAllEnabledClusters 计算所有集群数量
@@ -407,7 +411,7 @@ func (this *NodeClusterDAO) FindClusterDNSInfo(tx *dbs.Tx, clusterId int64, cach
if cacheMap == nil {
cacheMap = maps.Map{}
}
var cacheKey = this.Table + ":record:" + types.String(clusterId)
var cacheKey = this.Table + ":FindClusterDNSInfo:" + types.String(clusterId)
var cache = cacheMap.Get(cacheKey)
if cache != nil {
return cache.(*NodeCluster), nil
@@ -811,6 +815,28 @@ func (this *NodeClusterDAO) ExistsEnabledCluster(tx *dbs.Tx, clusterId int64) (b
Exist()
}
// FindClusterTimezone 查找时区
func (this *NodeClusterDAO) FindClusterTimezone(tx *dbs.Tx, clusterId int64, cacheMap maps.Map) (string, error) {
if cacheMap == nil {
cacheMap = maps.Map{}
}
var cacheKey = this.Table + ":FindEnabledTimeZone:" + types.String(clusterId)
var cache = cacheMap.Get(cacheKey)
if cache != nil {
return cache.(string), nil
}
timeZone, err := this.Query(tx).
Pk(clusterId).
Result("timeZone").
FindStringCol("")
if err != nil {
return "", err
}
cacheMap[cacheKey] = timeZone
return timeZone, nil
}
// NotifyUpdate 通知更新
func (this *NodeClusterDAO) NotifyUpdate(tx *dbs.Tx, clusterId int64) error {
return SharedNodeTaskDAO.CreateClusterTask(tx, nodeconfigs.NodeRoleNode, clusterId, NodeTaskTypeConfigChanged)

View File

@@ -26,6 +26,7 @@ type NodeCluster struct {
HttpFirewallPolicyId uint32 `field:"httpFirewallPolicyId"` // WAF策略ID
AccessLog string `field:"accessLog"` // 访问日志设置
SystemServices string `field:"systemServices"` // 系统服务设置
TimeZone string `field:"timeZone"` // 时区
}
type NodeClusterOperator struct {
@@ -53,6 +54,7 @@ type NodeClusterOperator struct {
HttpFirewallPolicyId interface{} // WAF策略ID
AccessLog interface{} // 访问日志设置
SystemServices interface{} // 系统服务设置
TimeZone interface{} // 时区
}
func NewNodeClusterOperator() *NodeClusterOperator {

View File

@@ -498,6 +498,22 @@ func (this *NodeDAO) FindAllEnabledNodesWithClusterId(tx *dbs.Tx, clusterId int6
return
}
// FindAllEnabledNodeIdsWithClusterId 获取一个集群的所有节点Ids
func (this *NodeDAO) FindAllEnabledNodeIdsWithClusterId(tx *dbs.Tx, clusterId int64) (result []int64, err error) {
ones, err := this.Query(tx).
ResultPk().
State(NodeStateEnabled).
Attr("clusterId", clusterId).
FindAll()
if err != nil {
return nil, err
}
for _, one := range ones {
result = append(result, int64(one.(*Node).Id))
}
return
}
// FindAllInactiveNodesWithClusterId 取得一个集群离线的节点
func (this *NodeDAO) FindAllInactiveNodesWithClusterId(tx *dbs.Tx, clusterId int64) (result []*Node, err error) {
_, err = this.Query(tx).
@@ -721,6 +737,10 @@ func (this *NodeDAO) ComposeNodeConfig(tx *dbs.Tx, nodeId int64, cacheMap maps.M
continue
}
config.Servers = append(config.Servers, serverConfig)
if server.IsOn == 1 && server.SupportCNAME == 1 {
config.SupportCNAME = true
}
}
// 全局设置
@@ -770,6 +790,15 @@ func (this *NodeDAO) ComposeNodeConfig(tx *dbs.Tx, nodeId int64, cacheMap maps.M
config.HTTPCachePolicies = append(config.HTTPCachePolicies, cachePolicy)
}
}
// 时区
timeZone, err := SharedNodeClusterDAO.FindClusterTimezone(tx, clusterId, cacheMap)
if err != nil {
return nil, err
}
if len(timeZone) > 0 {
config.TimeZone = timeZone
}
}
// 缓存最大容量设置

View File

@@ -75,6 +75,7 @@ func (this *NodeLogDAO) CreateLog(tx *dbs.Tx, nodeRole nodeconfigs.NodeRole, nod
op.Day = timeutil.FormatTime("Ymd", createdAt)
op.Hash = hash
op.Count = 1
op.IsRead = level != "error"
err = this.Save(tx, op)
return err
}
@@ -94,9 +95,20 @@ func (this *NodeLogDAO) DeleteExpiredLogs(tx *dbs.Tx, days int) error {
}
// CountNodeLogs 计算节点日志数量
func (this *NodeLogDAO) CountNodeLogs(tx *dbs.Tx, role string, nodeId int64, serverId int64, originId int64, dayFrom string, dayTo string, keyword string, level string) (int64, error) {
query := this.Query(tx).
Attr("role", role)
func (this *NodeLogDAO) CountNodeLogs(tx *dbs.Tx,
role string,
nodeId int64,
serverId int64,
originId int64,
dayFrom string,
dayTo string,
keyword string,
level string,
isUnread bool) (int64, error) {
query := this.Query(tx)
if len(role) > 0 {
query.Attr("role", role)
}
if nodeId > 0 {
query.Attr("nodeId", nodeId)
} else {
@@ -128,6 +140,9 @@ func (this *NodeLogDAO) CountNodeLogs(tx *dbs.Tx, role string, nodeId int64, ser
if len(level) > 0 {
query.Attr("level", level)
}
if isUnread {
query.Attr("isRead", 0)
}
return query.Count()
}
@@ -144,10 +159,13 @@ func (this *NodeLogDAO) ListNodeLogs(tx *dbs.Tx,
keyword string,
level string,
fixedState configutils.BoolState,
isUnread bool,
offset int64,
size int64) (result []*NodeLog, err error) {
query := this.Query(tx).
Attr("role", role)
query := this.Query(tx)
if len(role) > 0 {
query.Attr("role", role)
}
if nodeId > 0 {
query.Attr("nodeId", nodeId)
} else {
@@ -186,6 +204,9 @@ func (this *NodeLogDAO) ListNodeLogs(tx *dbs.Tx,
if len(level) > 0 {
query.Attr("level", level)
}
if isUnread {
query.Attr("isRead", 0)
}
_, err = query.
Offset(offset).
Limit(size).
@@ -224,3 +245,24 @@ func (this *NodeLogDAO) UpdateNodeLogFixed(tx *dbs.Tx, logId int64) error {
return nil
}
// CountAllUnreadNodeLogs 计算未读的日志数量
func (this *NodeLogDAO) CountAllUnreadNodeLogs(tx *dbs.Tx) (int64, error) {
return this.Query(tx).
Attr("isRead", false).
Count()
}
// UpdateNodeLogsRead 设置日志为已读
func (this *NodeLogDAO) UpdateNodeLogsRead(tx *dbs.Tx, nodeLogIds []int64) error {
for _, logId := range nodeLogIds {
err := this.Query(tx).
Pk(logId).
Set("isRead", true).
UpdateQuickly()
if err != nil {
return err
}
}
return nil
}

View File

@@ -15,6 +15,7 @@ type NodeLog struct {
Hash string `field:"hash"` // 信息内容Hash
Count uint32 `field:"count"` // 重复次数
IsFixed uint8 `field:"isFixed"` // 是否已处理
IsRead uint8 `field:"isRead"` // 是否已读
}
type NodeLogOperator struct {
@@ -31,6 +32,7 @@ type NodeLogOperator struct {
Hash interface{} // 信息内容Hash
Count interface{} // 重复次数
IsFixed interface{} // 是否已处理
IsRead interface{} // 是否已读
}
func NewNodeLogOperator() *NodeLogOperator {

View File

@@ -0,0 +1,179 @@
package models
import (
"github.com/TeaOSLab/EdgeAPI/internal/errors"
"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs"
_ "github.com/go-sql-driver/mysql"
"github.com/iwind/TeaGo/Tea"
"github.com/iwind/TeaGo/dbs"
)
const (
PlanStateEnabled = 1 // 已启用
PlanStateDisabled = 0 // 已禁用
)
type PlanDAO dbs.DAO
func NewPlanDAO() *PlanDAO {
return dbs.NewDAO(&PlanDAO{
DAOObject: dbs.DAOObject{
DB: Tea.Env,
Table: "edgePlans",
Model: new(Plan),
PkName: "id",
},
}).(*PlanDAO)
}
var SharedPlanDAO *PlanDAO
func init() {
dbs.OnReady(func() {
SharedPlanDAO = NewPlanDAO()
})
}
// EnablePlan 启用条目
func (this *PlanDAO) EnablePlan(tx *dbs.Tx, id uint32) error {
_, err := this.Query(tx).
Pk(id).
Set("state", PlanStateEnabled).
Update()
return err
}
// DisablePlan 禁用条目
func (this *PlanDAO) DisablePlan(tx *dbs.Tx, id int64) error {
_, err := this.Query(tx).
Pk(id).
Set("state", PlanStateDisabled).
Update()
return err
}
// FindEnabledPlan 查找启用中的条目
func (this *PlanDAO) FindEnabledPlan(tx *dbs.Tx, id int64) (*Plan, error) {
result, err := this.Query(tx).
Pk(id).
Attr("state", PlanStateEnabled).
Find()
if result == nil {
return nil, err
}
return result.(*Plan), err
}
// FindPlanName 根据主键查找名称
func (this *PlanDAO) FindPlanName(tx *dbs.Tx, id int64) (string, error) {
return this.Query(tx).
Pk(id).
Result("name").
FindStringCol("")
}
// CreatePlan 创建套餐
func (this *PlanDAO) CreatePlan(tx *dbs.Tx, name string, clusterId int64, bandwidthLimitJSON []byte, featuresJSON []byte, priceType serverconfigs.PlanPriceType, bandwidthPriceJSON []byte, monthlyPrice float32, seasonallyPrice float32, yearlyPrice float32) (int64, error) {
var op = NewPlanOperator()
op.Name = name
op.ClusterId = clusterId
if len(bandwidthLimitJSON) > 0 {
op.BandwidthLimit = bandwidthLimitJSON
}
if len(featuresJSON) > 0 {
op.Features = featuresJSON
}
op.PriceType = priceType
if len(bandwidthPriceJSON) > 0 {
op.BandwidthPrice = bandwidthPriceJSON
}
if monthlyPrice >= 0 {
op.MonthlyPrice = monthlyPrice
}
if seasonallyPrice >= 0 {
op.SeasonallyPrice = seasonallyPrice
}
if yearlyPrice >= 0 {
op.YearlyPrice = yearlyPrice
}
op.IsOn = true
op.State = PlanStateEnabled
return this.SaveInt64(tx, op)
}
// UpdatePlan 修改套餐
func (this *PlanDAO) UpdatePlan(tx *dbs.Tx, planId int64, name string, isOn bool, clusterId int64, bandwidthLimitJSON []byte, featuresJSON []byte, priceType serverconfigs.PlanPriceType, bandwidthPriceJSON []byte, monthlyPrice float32, seasonallyPrice float32, yearlyPrice float32) error {
if planId <= 0 {
return errors.New("invalid planId")
}
var op = NewPlanOperator()
op.Id = planId
op.Name = name
op.IsOn = isOn
op.ClusterId = clusterId
if len(bandwidthLimitJSON) > 0 {
op.BandwidthLimit = bandwidthLimitJSON
}
if len(featuresJSON) > 0 {
op.Features = featuresJSON
}
op.PriceType = priceType
if len(bandwidthPriceJSON) > 0 {
op.BandwidthPrice = bandwidthPriceJSON
}
if monthlyPrice >= 0 {
op.MonthlyPrice = monthlyPrice
} else {
op.MonthlyPrice = 0
}
if seasonallyPrice >= 0 {
op.SeasonallyPrice = seasonallyPrice
} else {
op.SeasonallyPrice = 0
}
if yearlyPrice >= 0 {
op.YearlyPrice = yearlyPrice
} else {
op.YearlyPrice = 0
}
return this.Save(tx, op)
}
// CountAllEnabledPlans 计算套餐的数量
func (this *PlanDAO) CountAllEnabledPlans(tx *dbs.Tx) (int64, error) {
return this.Query(tx).
State(PlanStateEnabled).
Count()
}
// ListEnabledPlans 列出单页套餐
func (this *PlanDAO) ListEnabledPlans(tx *dbs.Tx, offset int64, size int64) (result []*Plan, err error) {
_, err = this.Query(tx).
State(PlanStateEnabled).
Offset(offset).
Limit(size).
Slice(&result).
Desc("order").
DescPk().
FindAll()
return
}
// SortPlans 增加排序
func (this *PlanDAO) SortPlans(tx *dbs.Tx, planIds []int64) error {
if len(planIds) == 0 {
return nil
}
var order = len(planIds)
for _, planId := range planIds {
err := this.Query(tx).
Pk(planId).
Set("order", order).
UpdateQuickly()
if err != nil {
return err
}
order--
}
return nil
}

View File

@@ -0,0 +1,6 @@
package models
import (
_ "github.com/go-sql-driver/mysql"
_ "github.com/iwind/TeaGo/bootstrap"
)

View File

@@ -0,0 +1,38 @@
package models
// Plan 用户套餐
type Plan struct {
Id uint32 `field:"id"` // ID
IsOn uint8 `field:"isOn"` // 是否启用
Name string `field:"name"` // 套餐名
ClusterId uint32 `field:"clusterId"` // 集群ID
BandwidthLimit string `field:"bandwidthLimit"` // 带宽限制
Features string `field:"features"` // 允许的功能
BandwidthPrice string `field:"bandwidthPrice"` // 带宽价格设定
MonthlyPrice float64 `field:"monthlyPrice"` // 月付
SeasonallyPrice float64 `field:"seasonallyPrice"` // 季付
YearlyPrice float64 `field:"yearlyPrice"` // 年付
PriceType string `field:"priceType"` // 价格类型
Order uint32 `field:"order"` // 排序
State uint8 `field:"state"` // 状态
}
type PlanOperator struct {
Id interface{} // ID
IsOn interface{} // 是否启用
Name interface{} // 套餐名
ClusterId interface{} // 集群ID
BandwidthLimit interface{} // 带宽限制
Features interface{} // 允许的功能
BandwidthPrice interface{} // 带宽价格设定
MonthlyPrice interface{} // 月付
SeasonallyPrice interface{} // 季付
YearlyPrice interface{} // 年付
PriceType interface{} // 价格类型
Order interface{} // 排序
State interface{} // 状态
}
func NewPlanOperator() *PlanOperator {
return &PlanOperator{}
}

View File

@@ -0,0 +1 @@
package models

View File

@@ -190,6 +190,16 @@ func (this *ReverseProxyDAO) ComposeReverseProxyConfig(tx *dbs.Tx, reverseProxyI
config.IdleTimeout = idleTimeout
}
// PROXY Protocol
if IsNotNull(reverseProxy.ProxyProtocol) {
var proxyProtocolConfig = &serverconfigs.ProxyProtocolConfig{}
err = json.Unmarshal([]byte(reverseProxy.ProxyProtocol), proxyProtocolConfig)
if err != nil {
return nil, err
}
config.ProxyProtocol = proxyProtocolConfig
}
cacheMap[cacheKey] = config
return config, nil
@@ -286,7 +296,20 @@ func (this *ReverseProxyDAO) UpdateReverseProxyBackupOrigins(tx *dbs.Tx, reverse
}
// UpdateReverseProxy 修改是否启用
func (this *ReverseProxyDAO) UpdateReverseProxy(tx *dbs.Tx, reverseProxyId int64, requestHostType int8, requestHost string, requestURI string, stripPrefix string, autoFlush bool, addHeaders []string, connTimeout *shared.TimeDuration, readTimeout *shared.TimeDuration, idleTimeout *shared.TimeDuration, maxConns int32, maxIdleConns int32) error {
func (this *ReverseProxyDAO) UpdateReverseProxy(tx *dbs.Tx,
reverseProxyId int64,
requestHostType int8,
requestHost string,
requestURI string,
stripPrefix string,
autoFlush bool,
addHeaders []string,
connTimeout *shared.TimeDuration,
readTimeout *shared.TimeDuration,
idleTimeout *shared.TimeDuration,
maxConns int32,
maxIdleConns int32,
proxyProtocolJSON []byte) error {
if reverseProxyId <= 0 {
return errors.New("invalid reverseProxyId")
}
@@ -345,6 +368,10 @@ func (this *ReverseProxyDAO) UpdateReverseProxy(tx *dbs.Tx, reverseProxyId int64
op.MaxIdleConns = 0
}
if len(proxyProtocolJSON) > 0 {
op.ProxyProtocol = proxyProtocolJSON
}
err = this.Save(tx, op)
if err != nil {
return err

View File

@@ -1,6 +1,6 @@
package models
// 反向代理配置
// ReverseProxy 反向代理配置
type ReverseProxy struct {
Id uint32 `field:"id"` // ID
AdminId uint32 `field:"adminId"` // 管理员ID
@@ -23,6 +23,7 @@ type ReverseProxy struct {
IdleTimeout string `field:"idleTimeout"` // 空闲超时时间
MaxConns uint32 `field:"maxConns"` // 最大并发连接数
MaxIdleConns uint32 `field:"maxIdleConns"` // 最大空闲连接数
ProxyProtocol string `field:"proxyProtocol"` // Proxy Protocol配置
}
type ReverseProxyOperator struct {
@@ -47,6 +48,7 @@ type ReverseProxyOperator struct {
IdleTimeout interface{} // 空闲超时时间
MaxConns interface{} // 最大并发连接数
MaxIdleConns interface{} // 最大空闲连接数
ProxyProtocol interface{} // Proxy Protocol配置
}
func NewReverseProxyOperator() *ReverseProxyOperator {

View File

@@ -1,6 +1,7 @@
package models
import (
"github.com/TeaOSLab/EdgeAPI/internal/errors"
"github.com/TeaOSLab/EdgeAPI/internal/utils"
"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
_ "github.com/go-sql-driver/mysql"
@@ -53,12 +54,14 @@ func init() {
// SaveStats 提交数据
func (this *ServerDailyStatDAO) SaveStats(tx *dbs.Tx, stats []*pb.ServerDailyStat) error {
var serverUserMap = map[int64]int64{} // serverId => userId
var cacheMap = maps.Map{}
for _, stat := range stats {
day := timeutil.FormatTime("Ymd", stat.CreatedAt)
hour := timeutil.FormatTime("YmdH", stat.CreatedAt)
timeFrom := timeutil.FormatTime("His", stat.CreatedAt)
timeTo := timeutil.FormatTime("His", stat.CreatedAt+5*60-1) // 5分钟
// 所属用户
serverUserId, ok := serverUserMap[stat.ServerId]
if !ok {
userId, err := SharedServerDAO.FindServerUserId(tx, stat.ServerId)
@@ -100,7 +103,20 @@ func (this *ServerDailyStatDAO) SaveStats(tx *dbs.Tx, stats []*pb.ServerDailySta
if err != nil {
return err
}
// 更新带宽限制状态
bandwidthLimit, err := SharedServerDAO.FindServerBandwidthLimitConfig(tx, stat.ServerId, cacheMap)
if err != nil {
return err
}
if bandwidthLimit != nil && bandwidthLimit.IsOn && !bandwidthLimit.IsEmpty() {
err = SharedServerDAO.UpdateServerBandwidthLimitStatus(tx, bandwidthLimit, stat.ServerId, false)
if err != nil {
return err
}
}
}
return nil
}
@@ -233,7 +249,7 @@ func (this *ServerDailyStatDAO) SumDailyStat(tx *dbs.Tx, serverId int64, day str
stat = &pb.ServerDailyStat{}
if !regexp.MustCompile(`^\d{8}$`).MatchString(day) {
return
return nil, errors.New("invalid day '" + day + "'")
}
one, _, err := this.Query(tx).
@@ -258,6 +274,37 @@ func (this *ServerDailyStatDAO) SumDailyStat(tx *dbs.Tx, serverId int64, day str
return
}
// SumMonthlyStat 获取某月内的流量
// month 格式为YYYYMM
func (this *ServerDailyStatDAO) SumMonthlyStat(tx *dbs.Tx, serverId int64, month string) (stat *pb.ServerDailyStat, err error) {
stat = &pb.ServerDailyStat{}
if !regexp.MustCompile(`^\d{6}$`).MatchString(month) {
return
}
one, _, err := this.Query(tx).
Result("SUM(bytes) AS bytes, SUM(cachedBytes) AS cachedBytes, SUM(countRequests) AS countRequests, SUM(countCachedRequests) AS countCachedRequests, SUM(countAttackRequests) AS countAttackRequests, SUM(attackBytes) AS attackBytes").
Attr("serverId", serverId).
Between("day", month+"01", month+"31").
FindOne()
if err != nil {
return nil, err
}
if one == nil {
return
}
stat.Bytes = one.GetInt64("bytes")
stat.CachedBytes = one.GetInt64("cachedBytes")
stat.CountRequests = one.GetInt64("countRequests")
stat.CountCachedRequests = one.GetInt64("countCachedRequests")
stat.CountAttackRequests = one.GetInt64("countAttackRequests")
stat.AttackBytes = one.GetInt64("attackBytes")
return
}
// FindDailyStats 按天统计
func (this *ServerDailyStatDAO) FindDailyStats(tx *dbs.Tx, serverId int64, dayFrom string, dayTo string) (result []*ServerDailyStat, err error) {
ones, err := this.Query(tx).

View File

@@ -3,6 +3,7 @@ package models
import (
"encoding/json"
"errors"
"fmt"
"github.com/TeaOSLab/EdgeAPI/internal/db/models/dns"
"github.com/TeaOSLab/EdgeAPI/internal/utils/numberutils"
"github.com/TeaOSLab/EdgeCommon/pkg/configutils"
@@ -16,6 +17,7 @@ import (
"github.com/iwind/TeaGo/maps"
"github.com/iwind/TeaGo/rands"
"github.com/iwind/TeaGo/types"
timeutil "github.com/iwind/TeaGo/utils/time"
"regexp"
"strconv"
"strings"
@@ -50,7 +52,7 @@ func init() {
// Init 初始化
func (this *ServerDAO) Init() {
this.DAOObject.Init()
_ = this.DAOObject.Init()
// 这里不处理增删改事件是为了避免Server修改本身的时候也要触发别的Server变更
}
@@ -472,10 +474,26 @@ func (this *ServerDAO) UpdateServerWeb(tx *dbs.Tx, serverId int64, webId int64)
return this.NotifyUpdate(tx, serverId)
}
// UpdateServerDNS 修改DNS设置
func (this *ServerDAO) UpdateServerDNS(tx *dbs.Tx, serverId int64, supportCNAME bool) error {
if serverId <= 0 {
return errors.New("invalid serverId")
}
var op = NewServerOperator()
op.Id = serverId
op.SupportCNAME = supportCNAME
err := this.Save(tx, op)
if err != nil {
return err
}
return this.NotifyUpdate(tx, serverId)
}
// InitServerWeb 初始化Web配置
func (this *ServerDAO) InitServerWeb(tx *dbs.Tx, serverId int64) (int64, error) {
if serverId <= 0 {
return 0, errors.New("serverId should not be smaller than 0")
return 0, errors.New("invalid serverId")
}
adminId, userId, err := this.FindServerAdminIdAndUserId(tx, serverId)
@@ -859,6 +877,7 @@ func (this *ServerDAO) ComposeServerConfig(tx *dbs.Tx, server *Server, cacheMap
}
// CNAME
config.SupportCNAME = server.SupportCNAME == 1
if server.ClusterId > 0 && len(server.DnsName) > 0 {
clusterDNS, err := SharedNodeClusterDAO.FindClusterDNSInfo(tx, int64(server.ClusterId), cacheMap)
if err != nil {
@@ -1008,6 +1027,29 @@ func (this *ServerDAO) ComposeServerConfig(tx *dbs.Tx, server *Server, cacheMap
config.HTTPCachePolicyId = httpCachePolicyId
}
// bandwidth limit
if len(server.BandwidthLimit) > 0 {
var bandwidthLimitConfig = &serverconfigs.BandwidthLimitConfig{}
err = json.Unmarshal([]byte(server.BandwidthLimit), bandwidthLimitConfig)
if err != nil {
return nil, err
}
config.BandwidthLimit = bandwidthLimitConfig
if bandwidthLimitConfig.IsOn && !bandwidthLimitConfig.IsEmpty() {
if len(server.BandwidthLimitStatus) > 0 {
var status = &serverconfigs.BandwidthLimitStatus{}
err = json.Unmarshal([]byte(server.BandwidthLimitStatus), status)
if err != nil {
return nil, err
}
if status.IsValid() {
config.BandwidthLimitStatus = status
}
}
}
}
return config, nil
}
@@ -1190,6 +1232,54 @@ func (this *ServerDAO) FindAllServersDNSWithClusterId(tx *dbs.Tx, clusterId int6
return
}
// FindAllEnabledServersWithDomain 根据域名查找服务
func (this *ServerDAO) FindAllEnabledServersWithDomain(tx *dbs.Tx, domain string) (result []*Server, err error) {
if len(domain) == 0 {
return
}
_, err = this.Query(tx).
State(ServerStateEnabled).
Where("(JSON_CONTAINS(serverNames, :domain1) OR JSON_CONTAINS(serverNames, :domain2))").
Param("domain1", maps.Map{"name": domain}.AsJSON()).
Param("domain2", maps.Map{"subNames": domain}.AsJSON()).
Slice(&result).
DescPk().
FindAll()
if err != nil {
return nil, err
}
// 支持泛解析
var countPieces = strings.Count(domain, ".")
for {
var index = strings.Index(domain, ".")
if index > 0 {
domain = domain[index+1:]
var search = strings.Repeat("*.", countPieces-strings.Count(domain, ".")) + domain
_, err = this.Query(tx).
State(ServerStateEnabled).
Where("(JSON_CONTAINS(serverNames, :domain1) OR JSON_CONTAINS(serverNames, :domain2))").
Param("domain1", maps.Map{"name": search}.AsJSON()).
Param("domain2", maps.Map{"subNames": search}.AsJSON()).
Slice(&result).
DescPk().
FindAll()
if err != nil {
return
}
if len(result) > 0 {
return
}
} else {
break
}
}
return
}
// GenerateServerDNSName 重新生成子域名
func (this *ServerDAO) GenerateServerDNSName(tx *dbs.Tx, serverId int64) (string, error) {
if serverId <= 0 {
@@ -1236,6 +1326,18 @@ func (this *ServerDAO) FindServerDNSName(tx *dbs.Tx, serverId int64) (string, er
FindStringCol("")
}
// FindServerSupportCNAME 查询服务是否支持CNAME
func (this *ServerDAO) FindServerSupportCNAME(tx *dbs.Tx, serverId int64) (bool, error) {
supportCNAME, err := this.Query(tx).
Pk(serverId).
Result("supportCNAME").
FindIntCol(0)
if err != nil {
return false, err
}
return supportCNAME == 1, nil
}
// FindStatelessServerDNS 查询服务的DNS相关信息并且不关注状态
func (this *ServerDAO) FindStatelessServerDNS(tx *dbs.Tx, serverId int64) (*Server, error) {
one, err := this.Query(tx).
@@ -1661,6 +1763,170 @@ func (this *ServerDAO) NotifyServerPortsUpdate(tx *dbs.Tx, serverId int64) error
UpdateQuickly()
}
// FindServerBandwidthLimitConfig 查找服务的带宽限制
func (this *ServerDAO) FindServerBandwidthLimitConfig(tx *dbs.Tx, serverId int64, cacheMap maps.Map) (*serverconfigs.BandwidthLimitConfig, error) {
if cacheMap == nil {
cacheMap = maps.Map{}
}
var cacheKey = this.Table + ":FindServerBandwidthLimitConfig:" + types.String(serverId)
result, ok := cacheMap[cacheKey]
if ok {
return result.(*serverconfigs.BandwidthLimitConfig), nil
}
bandwidthLimit, err := this.Query(tx).
Pk(serverId).
Result("bandwidthLimit").
FindStringCol("")
if err != nil {
return nil, err
}
var limit = &serverconfigs.BandwidthLimitConfig{}
if len(bandwidthLimit) == 0 {
return limit, nil
}
err = json.Unmarshal([]byte(bandwidthLimit), limit)
if err != nil {
return nil, err
}
cacheMap[cacheKey] = limit
return limit, nil
}
// UpdateServerBandwidthLimitConfig 修改服务的带宽限制
func (this *ServerDAO) UpdateServerBandwidthLimitConfig(tx *dbs.Tx, serverId int64, bandwidthLimitConfig *serverconfigs.BandwidthLimitConfig) error {
if serverId <= 0 {
return errors.New("invalid serverId")
}
limitJSON, err := json.Marshal(bandwidthLimitConfig)
if err != nil {
return err
}
err = this.Query(tx).
Pk(serverId).
Set("bandwidthLimit", limitJSON).
UpdateQuickly()
if err != nil {
return err
}
// 更新状态
return this.UpdateServerBandwidthLimitStatus(tx, bandwidthLimitConfig, serverId, true)
}
func (this *ServerDAO) UpdateServerBandwidthLimitStatus(tx *dbs.Tx, bandwidthLimitConfig *serverconfigs.BandwidthLimitConfig, serverId int64, isUpdatingConfig bool) error {
if !bandwidthLimitConfig.IsOn {
if isUpdatingConfig {
return this.NotifyUpdate(tx, serverId)
}
return nil
}
oldStatusString, err := this.Query(tx).
Pk(serverId).
Result("bandwidthLimitStatus").
FindStringCol("")
if err != nil {
return err
}
var oldStatus = &serverconfigs.BandwidthLimitStatus{}
if len(oldStatusString) > 0 {
err = json.Unmarshal([]byte(oldStatusString), oldStatus)
if err != nil {
return err
}
// 如果已经达到限制了,而且还在有效期,那就没必要再更新
if !isUpdatingConfig && oldStatus.IsValid() {
return nil
}
}
var untilDay = ""
// daily
if bandwidthLimitConfig.DailyBytes() > 0 {
stat, err := SharedServerDailyStatDAO.SumDailyStat(tx, serverId, timeutil.Format("Ymd"))
if err != nil {
return err
}
if stat != nil && stat.Bytes >= bandwidthLimitConfig.DailyBytes() {
untilDay = timeutil.Format("Ymd")
}
}
// monthly
if bandwidthLimitConfig.MonthlyBytes() > 0 {
stat, err := SharedServerDailyStatDAO.SumMonthlyStat(tx, serverId, timeutil.Format("Ym"))
if err != nil {
return err
}
if stat != nil && stat.Bytes >= bandwidthLimitConfig.MonthlyBytes() {
untilDay = timeutil.Format("Ym") + fmt.Sprintf("%02d", types.Int(timeutil.Format("t")))
}
}
// totally
if bandwidthLimitConfig.TotalBytes() > 0 {
totalBandwidth, err := this.Query(tx).
Pk(serverId).
Result("totalBandwidth").
FindFloat64Col(0)
if err != nil {
return err
}
if totalBandwidth >= float64(bandwidthLimitConfig.TotalBytes()) {
untilDay = "20990101"
}
}
var isChanged = oldStatus.UntilDay != untilDay
if isChanged {
statusJSON, err := json.Marshal(&serverconfigs.BandwidthLimitStatus{UntilDay: untilDay})
if err != nil {
return err
}
err = this.Query(tx).
Pk(serverId).
Set("bandwidthLimitStatus", statusJSON).
UpdateQuickly()
if err != nil {
return err
}
return this.NotifyUpdate(tx, serverId)
}
if isUpdatingConfig {
return this.NotifyUpdate(tx, serverId)
}
return nil
}
// IncreaseServerTotalBandwidth 增加服务的总带宽
func (this *ServerDAO) IncreaseServerTotalBandwidth(tx *dbs.Tx, serverId int64, bytes int64) error {
var gb = float64(bytes) / 1024 / 1024 / 1024
return this.Query(tx).
Pk(serverId).
Set("totalBandwidth", dbs.SQL("totalBandwidth+:bandwidthGB")).
Param("bandwidthGB", gb).
UpdateQuickly()
}
// ResetServerTotalBandwidth 重置服务总带宽
func (this *ServerDAO) ResetServerTotalBandwidth(tx *dbs.Tx, serverId int64) error {
return this.Query(tx).
Pk(serverId).
Set("totalBandwidth", 0).
UpdateQuickly()
}
// NotifyUpdate 同步集群
func (this *ServerDAO) NotifyUpdate(tx *dbs.Tx, serverId int64) error {
// 创建任务

View File

@@ -3,10 +3,13 @@ package models
import (
"crypto/md5"
"encoding/json"
"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs"
"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs/shared"
_ "github.com/go-sql-driver/mysql"
"github.com/iwind/TeaGo/dbs"
"github.com/iwind/TeaGo/logs"
"testing"
"time"
)
func TestServerDAO_ComposeServerConfig(t *testing.T) {
@@ -145,6 +148,43 @@ func TestServerDAO_FindAllEnabledServersWithNode(t *testing.T) {
}
}
func TestServerDAO_FindAllEnabledServersWithDomain(t *testing.T) {
for _, domain := range []string{"yun4s.cn", "teaos.cn", "teaos2.cn", "cdn.teaos.cn", "cdn100.teaos.cn"} {
servers, err := NewServerDAO().FindAllEnabledServersWithDomain(nil, domain)
if err != nil {
t.Fatal(err)
}
if len(servers) > 0 {
for _, server := range servers {
t.Log(domain + ": " + server.ServerNames)
}
} else {
t.Log(domain + ": not found")
}
}
}
func TestServerDAO_UpdateServerBandwidthLimitStatus(t *testing.T) {
dbs.NotifyReady()
var tx *dbs.Tx
before := time.Now()
defer func() {
t.Log(time.Since(before).Seconds()*1000, "ms")
}()
err := NewServerDAO().UpdateServerBandwidthLimitStatus(tx, &serverconfigs.BandwidthLimitConfig{
IsOn: true,
DailySize: &shared.SizeCapacity{Count: 1, Unit: "mb"},
MonthlySize: &shared.SizeCapacity{Count: 10, Unit: "mb"},
TotalSize: nil,
NoticePageBody: "",
}, 23, false)
if err != nil {
t.Fatal(err)
}
t.Log("ok")
}
func BenchmarkServerDAO_CountAllEnabledServers(b *testing.B) {
SharedServerDAO = NewServerDAO()

View File

@@ -2,71 +2,81 @@ package models
// Server 服务
type Server struct {
Id uint32 `field:"id"` // ID
IsOn uint8 `field:"isOn"` // 是否启用
UserId uint32 `field:"userId"` // 用户ID
AdminId uint32 `field:"adminId"` // 管理员ID
Type string `field:"type"` // 服务类型
Name string `field:"name"` // 名称
Description string `field:"description"` // 描述
ServerNames string `field:"serverNames"` // 域名列表
AuditingServerNames string `field:"auditingServerNames"` // 审核中的域名
IsAuditing uint8 `field:"isAuditing"` // 是否正在审核
AuditingResult string `field:"auditingResult"` // 审核结果
Http string `field:"http"` // HTTP配置
Https string `field:"https"` // HTTPS配置
Tcp string `field:"tcp"` // TCP配置
Tls string `field:"tls"` // TLS配置
Unix string `field:"unix"` // Unix配置
Udp string `field:"udp"` // UDP配置
WebId uint32 `field:"webId"` // WEB配置
ReverseProxy string `field:"reverseProxy"` // 反向代理配置
GroupIds string `field:"groupIds"` // 分组ID列表
Config string `field:"config"` // 服务配置,自动生成
ConfigMd5 string `field:"configMd5"` // Md5
ClusterId uint32 `field:"clusterId"` // 集群ID
IncludeNodes string `field:"includeNodes"` // 部署条件
ExcludeNodes string `field:"excludeNodes"` // 节点排除条件
Version uint32 `field:"version"` // 版本号
CreatedAt uint64 `field:"createdAt"` // 创建时间
State uint8 `field:"state"` // 状态
DnsName string `field:"dnsName"` // DNS名称
TcpPorts string `field:"tcpPorts"` // 所包含TCP端口
UdpPorts string `field:"udpPorts"` // 所包含UDP端口
Id uint32 `field:"id"` // ID
IsOn uint8 `field:"isOn"` // 是否启用
UserId uint32 `field:"userId"` // 用户ID
AdminId uint32 `field:"adminId"` // 管理员ID
Type string `field:"type"` // 服务类型
Name string `field:"name"` // 名称
Description string `field:"description"` // 描述
ServerNames string `field:"serverNames"` // 域名列表
AuditingServerNames string `field:"auditingServerNames"` // 审核中的域名
IsAuditing uint8 `field:"isAuditing"` // 是否正在审核
AuditingResult string `field:"auditingResult"` // 审核结果
Http string `field:"http"` // HTTP配置
Https string `field:"https"` // HTTPS配置
Tcp string `field:"tcp"` // TCP配置
Tls string `field:"tls"` // TLS配置
Unix string `field:"unix"` // Unix配置
Udp string `field:"udp"` // UDP配置
WebId uint32 `field:"webId"` // WEB配置
ReverseProxy string `field:"reverseProxy"` // 反向代理配置
GroupIds string `field:"groupIds"` // 分组ID列表
Config string `field:"config"` // 服务配置,自动生成
ConfigMd5 string `field:"configMd5"` // Md5
ClusterId uint32 `field:"clusterId"` // 集群ID
IncludeNodes string `field:"includeNodes"` // 部署条件
ExcludeNodes string `field:"excludeNodes"` // 节点排除条件
Version uint32 `field:"version"` // 版本号
CreatedAt uint64 `field:"createdAt"` // 创建时间
State uint8 `field:"state"` // 状态
DnsName string `field:"dnsName"` // DNS名称
TcpPorts string `field:"tcpPorts"` // 所包含TCP端口
UdpPorts string `field:"udpPorts"` // 所包含UDP端口
SupportCNAME uint8 `field:"supportCNAME"` // 允许CNAME不在域名名单
BandwidthLimit string `field:"bandwidthLimit"` // 带宽限制
TotalBandwidth float64 `field:"totalBandwidth"` // 总带宽用量单位GB
BandwidthLimitStatus string `field:"bandwidthLimitStatus"` // 带宽限制状态
UserPlanId uint32 `field:"userPlanId"` // 所属套餐ID
}
type ServerOperator struct {
Id interface{} // ID
IsOn interface{} // 是否启用
UserId interface{} // 用户ID
AdminId interface{} // 管理员ID
Type interface{} // 服务类型
Name interface{} // 名称
Description interface{} // 描述
ServerNames interface{} // 域名列表
AuditingServerNames interface{} // 审核中的域名
IsAuditing interface{} // 是否正在审核
AuditingResult interface{} // 审核结果
Http interface{} // HTTP配置
Https interface{} // HTTPS配置
Tcp interface{} // TCP配置
Tls interface{} // TLS配置
Unix interface{} // Unix配置
Udp interface{} // UDP配置
WebId interface{} // WEB配置
ReverseProxy interface{} // 反向代理配置
GroupIds interface{} // 分组ID列表
Config interface{} // 服务配置,自动生成
ConfigMd5 interface{} // Md5
ClusterId interface{} // 集群ID
IncludeNodes interface{} // 部署条件
ExcludeNodes interface{} // 节点排除条件
Version interface{} // 版本号
CreatedAt interface{} // 创建时间
State interface{} // 状态
DnsName interface{} // DNS名称
TcpPorts interface{} // 所包含TCP端口
UdpPorts interface{} // 所包含UDP端口
Id interface{} // ID
IsOn interface{} // 是否启用
UserId interface{} // 用户ID
AdminId interface{} // 管理员ID
Type interface{} // 服务类型
Name interface{} // 名称
Description interface{} // 描述
ServerNames interface{} // 域名列表
AuditingServerNames interface{} // 审核中的域名
IsAuditing interface{} // 是否正在审核
AuditingResult interface{} // 审核结果
Http interface{} // HTTP配置
Https interface{} // HTTPS配置
Tcp interface{} // TCP配置
Tls interface{} // TLS配置
Unix interface{} // Unix配置
Udp interface{} // UDP配置
WebId interface{} // WEB配置
ReverseProxy interface{} // 反向代理配置
GroupIds interface{} // 分组ID列表
Config interface{} // 服务配置,自动生成
ConfigMd5 interface{} // Md5
ClusterId interface{} // 集群ID
IncludeNodes interface{} // 部署条件
ExcludeNodes interface{} // 节点排除条件
Version interface{} // 版本号
CreatedAt interface{} // 创建时间
State interface{} // 状态
DnsName interface{} // DNS名称
TcpPorts interface{} // 所包含TCP端口
UdpPorts interface{} // 所包含UDP端口
SupportCNAME interface{} // 允许CNAME不在域名名单
BandwidthLimit interface{} // 带宽限制
TotalBandwidth interface{} // 总带宽用量单位GB
BandwidthLimitStatus interface{} // 带宽限制状态
UserPlanId interface{} // 所属套餐ID
}
func NewServerOperator() *ServerOperator {

View File

@@ -24,7 +24,7 @@ func init() {
var ticker = time.NewTicker(time.Duration(rands.Int(24, 48)) * time.Hour)
go func() {
for range ticker.C {
err := SharedServerDomainHourlyStatDAO.Clean(nil, 60) // 只保留60
err := SharedServerDomainHourlyStatDAO.Clean(nil, 7) // 只保留7
if err != nil {
remotelogs.Error("ServerDomainHourlyStatDAO", "clean expired data failed: "+err.Error())
}

View File

@@ -0,0 +1,138 @@
package models
import (
"github.com/TeaOSLab/EdgeAPI/internal/errors"
_ "github.com/go-sql-driver/mysql"
"github.com/iwind/TeaGo/Tea"
"github.com/iwind/TeaGo/dbs"
timeutil "github.com/iwind/TeaGo/utils/time"
"time"
)
const (
UserPlanStateEnabled = 1 // 已启用
UserPlanStateDisabled = 0 // 已禁用
)
type UserPlanDAO dbs.DAO
func NewUserPlanDAO() *UserPlanDAO {
return dbs.NewDAO(&UserPlanDAO{
DAOObject: dbs.DAOObject{
DB: Tea.Env,
Table: "edgeUserPlans",
Model: new(UserPlan),
PkName: "id",
},
}).(*UserPlanDAO)
}
var SharedUserPlanDAO *UserPlanDAO
func init() {
dbs.OnReady(func() {
SharedUserPlanDAO = NewUserPlanDAO()
})
}
// EnableUserPlan 启用条目
func (this *UserPlanDAO) EnableUserPlan(tx *dbs.Tx, id uint64) error {
_, err := this.Query(tx).
Pk(id).
Set("state", UserPlanStateEnabled).
Update()
return err
}
// DisableUserPlan 禁用条目
func (this *UserPlanDAO) DisableUserPlan(tx *dbs.Tx, id int64) error {
_, err := this.Query(tx).
Pk(id).
Set("state", UserPlanStateDisabled).
Update()
return err
}
// FindEnabledUserPlan 查找启用中的条目
func (this *UserPlanDAO) FindEnabledUserPlan(tx *dbs.Tx, id int64) (*UserPlan, error) {
result, err := this.Query(tx).
Pk(id).
Attr("state", UserPlanStateEnabled).
Find()
if result == nil {
return nil, err
}
return result.(*UserPlan), err
}
// CountAllEnabledUserPlans 计算套餐数量
func (this *UserPlanDAO) CountAllEnabledUserPlans(tx *dbs.Tx, isAvailable bool, isExpired bool, expiringDays int32) (int64, error) {
var query = this.Query(tx).
State(UserPlanStateEnabled).
Where("userId IN (SELECT id FROM " + SharedUserDAO.Table + " WHERE state=1)").
Where("planId IN (SELECT id FROM " + SharedPlanDAO.Table + " WHERE state=1)")
var today = timeutil.Format("Y-m-d")
if isAvailable {
query.Gte("dayTo", today)
}
if isExpired {
query.Lt("dayTo", today)
}
if expiringDays > 0 {
var expiringDay = timeutil.Format("Y-m-d", time.Now().AddDate(0, 0, int(expiringDays)))
query.Gte("dayTo", today)
query.Lte("dayTo", expiringDay)
}
return query.Count()
}
// ListEnabledUserPlans 列出单页套餐
func (this *UserPlanDAO) ListEnabledUserPlans(tx *dbs.Tx, isAvailable bool, isExpired bool, expiringDays int32, offset int64, size int64) (result []*UserPlan, err error) {
var query = this.Query(tx).
State(UserPlanStateEnabled).
Where("userId IN (SELECT id FROM " + SharedUserDAO.Table + " WHERE state=1)").
Where("planId IN (SELECT id FROM " + SharedPlanDAO.Table + " WHERE state=1)")
var today = timeutil.Format("Y-m-d")
if isAvailable {
query.Gte("dayTo", today)
}
if isExpired {
query.Lt("dayTo", today)
}
if expiringDays > 0 {
var expiringDay = timeutil.Format("Y-m-d", time.Now().AddDate(0, 0, int(expiringDays)))
query.Gte("dayTo", today)
query.Lte("dayTo", expiringDay)
}
_, err = query.
DescPk().
Offset(offset).
Limit(size).
Slice(&result).
FindAll()
return
}
// CreateUserPlan 创建套餐
func (this *UserPlanDAO) CreateUserPlan(tx *dbs.Tx, userId int64, planId int64, dayTo string) (int64, error) {
var op = NewUserPlanOperator()
op.UserId = userId
op.PlanId = planId
op.DayTo = dayTo
op.IsOn = true
op.State = UserStateEnabled
return this.SaveInt64(tx, op)
}
// UpdateUserPlan 修改套餐
func (this *UserPlanDAO) UpdateUserPlan(tx *dbs.Tx, userPlanId int64, planId int64, dayTo string, isOn bool) error {
if userPlanId <= 0 {
return errors.New("invalid userPlanId")
}
var op = NewUserPlanOperator()
op.Id = userPlanId
op.PlanId = planId
op.DayTo = dayTo
op.IsOn = isOn
return this.Save(tx, op)
}

View File

@@ -0,0 +1,6 @@
package models
import (
_ "github.com/go-sql-driver/mysql"
_ "github.com/iwind/TeaGo/bootstrap"
)

View File

@@ -0,0 +1,24 @@
package models
// UserPlan 用户的套餐
type UserPlan struct {
Id uint64 `field:"id"` // ID
UserId uint32 `field:"userId"` // 用户ID
PlanId uint32 `field:"planId"` // 套餐ID
IsOn uint8 `field:"isOn"` // 是否启用
DayTo string `field:"dayTo"` // 结束日期
State uint8 `field:"state"` // 状态
}
type UserPlanOperator struct {
Id interface{} // ID
UserId interface{} // 用户ID
PlanId interface{} // 套餐ID
IsOn interface{} // 是否启用
DayTo interface{} // 结束日期
State interface{} // 状态
}
func NewUserPlanOperator() *UserPlanOperator {
return &UserPlanOperator{}
}

View File

@@ -0,0 +1 @@
package models

View File

@@ -1339,6 +1339,12 @@ func (this *HuaweiDNSProvider) AddRecord(domain string, newRecord *dnstypes.Reco
if ttl <= 0 {
ttl = 300
}
// 华为云TXT需要加引号
if newRecord.Type == dnstypes.RecordTypeTXT {
newRecord.Value = "\"" + strings.Trim(newRecord.Value, "\"") + "\""
}
err = this.doAPI(http.MethodPost, "/v2.1/zones/"+zoneId+"/recordsets", map[string]string{}, maps.Map{
"name": newRecord.Name + "." + domain + ".",
"description": "CDN系统自动创建",
@@ -1376,6 +1382,11 @@ func (this *HuaweiDNSProvider) UpdateRecord(domain string, record *dnstypes.Reco
ttl = 300
}
// 华为云TXT需要加引号
if newRecord.Type == dnstypes.RecordTypeTXT {
newRecord.Value = "\"" + strings.Trim(newRecord.Value, "\"") + "\""
}
var resp = new(huaweidns.ZonesUpdateRecordSetResponse)
err = this.doAPI(http.MethodPut, "/v2.1/zones/"+zoneId+"/recordsets/"+recordId, map[string]string{}, maps.Map{
"name": newRecord.Name + "." + domain + ".",

View File

@@ -526,6 +526,18 @@ func (this *APINode) registerServices(server *grpc.Server) {
this.rest(instance)
}
{
instance := this.serviceInstance(&services.PlanService{}).(*services.PlanService)
pb.RegisterPlanServiceServer(server, instance)
this.rest(instance)
}
{
instance := this.serviceInstance(&services.UserPlanService{}).(*services.UserPlanService)
pb.RegisterUserPlanServiceServer(server, instance)
this.rest(instance)
}
APINodeServicesRegister(this, server)
// TODO check service names

View File

@@ -45,7 +45,13 @@ func (this *RestServer) handle(writer http.ResponseWriter, req *http.Request) {
path := req.URL.Path
// 是否显示Pretty后的JSON
shouldPretty := req.Header.Get("Edge-Response-Pretty") == "on"
shouldPretty := req.Header.Get("X-Edge-Response-Pretty") == "on"
// 兼容老的Header
var oldShouldPretty = req.Header.Get("Edge-Response-Pretty")
if len(oldShouldPretty) > 0 {
shouldPretty = oldShouldPretty == "on"
}
// 欢迎页
if path == "/" {
@@ -98,14 +104,17 @@ func (this *RestServer) handle(writer http.ResponseWriter, req *http.Request) {
if serviceName != "APIAccessTokenService" || (methodName != "GetAPIAccessToken" && methodName != "getAPIAccessToken") {
// 校验TOKEN
token := req.Header.Get("Edge-Access-Token")
token := req.Header.Get("X-Edge-Access-Token")
if len(token) == 0 {
this.writeJSON(writer, maps.Map{
"code": 400,
"data": maps.Map{},
"message": "require 'Edge-Access-Token' header",
}, shouldPretty)
return
token = req.Header.Get("Edge-Access-Token")
if len(token) == 0 {
this.writeJSON(writer, maps.Map{
"code": 400,
"data": maps.Map{},
"message": "require 'X-Edge-Access-Token' header",
}, shouldPretty)
return
}
}
accessToken, err := models.SharedAPIAccessTokenDAO.FindAccessToken(nil, token)

View File

@@ -4,7 +4,9 @@ import (
"context"
"encoding/json"
"github.com/TeaOSLab/EdgeAPI/internal/db/models"
"github.com/TeaOSLab/EdgeAPI/internal/errors"
"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs/firewallconfigs"
)
// HTTPFirewallRuleGroupService WAF规则分组相关服务
@@ -48,7 +50,7 @@ func (this *HTTPFirewallRuleGroupService) CreateHTTPFirewallRuleGroup(ctx contex
tx := this.NullTx()
groupId, err := models.SharedHTTPFirewallRuleGroupDAO.CreateGroup(tx, req.IsOn, req.Name, req.Description)
groupId, err := models.SharedHTTPFirewallRuleGroupDAO.CreateGroup(tx, req.IsOn, req.Name, req.Code, req.Description)
if err != nil {
return nil, err
}
@@ -167,7 +169,7 @@ func (this *HTTPFirewallRuleGroupService) UpdateHTTPFirewallRuleGroupSets(ctx co
return nil, err
}
}
tx := this.NullTx()
err = models.SharedHTTPFirewallRuleGroupDAO.UpdateGroupSets(tx, req.GetFirewallRuleGroupId(), req.FirewallRuleSetsJSON)
@@ -176,3 +178,65 @@ func (this *HTTPFirewallRuleGroupService) UpdateHTTPFirewallRuleGroupSets(ctx co
}
return this.Success()
}
// AddHTTPFirewallRuleGroupSet 添加规则集
func (this *HTTPFirewallRuleGroupService) AddHTTPFirewallRuleGroupSet(ctx context.Context, req *pb.AddHTTPFirewallRuleGroupSetRequest) (*pb.RPCSuccess, error) {
// 校验请求
_, userId, err := this.ValidateAdminAndUser(ctx, 0, 0)
if err != nil {
return nil, err
}
if userId > 0 {
// 校验权限
err = models.SharedHTTPFirewallRuleGroupDAO.CheckUserRuleGroup(nil, userId, req.FirewallRuleGroupId)
if err != nil {
return nil, err
}
}
tx := this.NullTx()
// 已经有的规则
config, err := models.SharedHTTPFirewallRuleGroupDAO.ComposeFirewallRuleGroup(tx, req.FirewallRuleGroupId)
if err != nil {
return nil, err
}
if config == nil {
return nil, errors.New("can not find group")
}
var setRefs = config.SetRefs
var set = &firewallconfigs.HTTPFirewallRuleSet{}
err = json.Unmarshal(req.FirewallRuleSetConfigJSON, set)
if err != nil {
return nil, err
}
if set.Id > 0 {
setRefs = append(setRefs, &firewallconfigs.HTTPFirewallRuleSetRef{
IsOn: true,
SetId: set.Id,
})
} else {
setId, err := models.SharedHTTPFirewallRuleSetDAO.CreateOrUpdateSetFromConfig(tx, set)
if err != nil {
return nil, err
}
setRefs = append(setRefs, &firewallconfigs.HTTPFirewallRuleSetRef{
IsOn: true,
SetId: setId,
})
}
setRefsJSON, err := json.Marshal(setRefs)
if err != nil {
return nil, err
}
err = models.SharedHTTPFirewallRuleGroupDAO.UpdateGroupSets(tx, req.FirewallRuleGroupId, setRefsJSON)
if err != nil {
return nil, err
}
return this.Success()
}

View File

@@ -151,7 +151,7 @@ func (this *IPItemService) CountIPItemsWithListId(ctx context.Context, req *pb.C
}
}
count, err := models.SharedIPItemDAO.CountIPItemsWithListId(tx, req.IpListId)
count, err := models.SharedIPItemDAO.CountIPItemsWithListId(tx, req.IpListId, req.Keyword, req.IpFrom, req.IpTo)
if err != nil {
return nil, err
}
@@ -175,7 +175,7 @@ func (this *IPItemService) ListIPItemsWithListId(ctx context.Context, req *pb.Li
}
}
items, err := models.SharedIPItemDAO.ListIPItemsWithListId(tx, req.IpListId, req.Offset, req.Size)
items, err := models.SharedIPItemDAO.ListIPItemsWithListId(tx, req.IpListId, req.Keyword, req.IpFrom, req.IpTo, req.Offset, req.Size)
if err != nil {
return nil, err
}

View File

@@ -1328,6 +1328,7 @@ func (this *NodeService) UpdateNodeDNS(ctx context.Context, req *pb.UpdateNodeDN
delete(routeCodeMap, req.DnsDomainId)
}
} else {
routeCodeMap = map[int64][]string{}
if len(req.Routes) > 0 {
var m = map[int64][]string{} // domainId => codes
for _, route := range req.Routes {
@@ -1342,9 +1343,6 @@ func (this *NodeService) UpdateNodeDNS(ctx context.Context, req *pb.UpdateNodeDN
for domainId, codes := range m {
routeCodeMap[domainId] = codes
}
} else {
// 清空
routeCodeMap = map[int64][]string{}
}
}

View File

@@ -83,7 +83,7 @@ func (this *NodeClusterService) UpdateNodeCluster(ctx context.Context, req *pb.U
tx := this.NullTx()
err = models.SharedNodeClusterDAO.UpdateCluster(tx, req.NodeClusterId, req.Name, req.NodeGrantId, req.InstallDir)
err = models.SharedNodeClusterDAO.UpdateCluster(tx, req.NodeClusterId, req.Name, req.NodeGrantId, req.InstallDir, req.TimeZone)
if err != nil {
return nil, err
}
@@ -160,6 +160,7 @@ func (this *NodeClusterService) FindEnabledNodeCluster(ctx context.Context, req
DnsName: cluster.DnsName,
DnsDomainId: int64(cluster.DnsDomainId),
IsOn: cluster.IsOn == 1,
TimeZone: cluster.TimeZone,
}}, nil
}
@@ -293,6 +294,7 @@ func (this *NodeClusterService) ListEnabledNodeClusters(ctx context.Context, req
DnsName: cluster.DnsName,
DnsDomainId: int64(cluster.DnsDomainId),
IsOn: cluster.IsOn == 1,
TimeZone: cluster.TimeZone,
})
}

View File

@@ -41,7 +41,7 @@ func (this *NodeLogService) CountNodeLogs(ctx context.Context, req *pb.CountNode
tx := this.NullTx()
count, err := models.SharedNodeLogDAO.CountNodeLogs(tx, req.Role, req.NodeId, req.ServerId, req.OriginId, req.DayFrom, req.DayTo, req.Keyword, req.Level)
count, err := models.SharedNodeLogDAO.CountNodeLogs(tx, req.Role, req.NodeId, req.ServerId, req.OriginId, req.DayFrom, req.DayTo, req.Keyword, req.Level, req.IsUnread)
if err != nil {
return nil, err
}
@@ -57,7 +57,7 @@ func (this *NodeLogService) ListNodeLogs(ctx context.Context, req *pb.ListNodeLo
tx := this.NullTx()
logs, err := models.SharedNodeLogDAO.ListNodeLogs(tx, req.Role, req.NodeId, req.ServerId, req.OriginId, req.AllServers, req.DayFrom, req.DayTo, req.Keyword, req.Level, types.Int8(req.FixedState), req.Offset, req.Size)
logs, err := models.SharedNodeLogDAO.ListNodeLogs(tx, req.Role, req.NodeId, req.ServerId, req.OriginId, req.AllServers, req.DayFrom, req.DayTo, req.Keyword, req.Level, types.Int8(req.FixedState), req.IsUnread, req.Offset, req.Size)
if err != nil {
return nil, err
}
@@ -85,6 +85,7 @@ func (this *NodeLogService) ListNodeLogs(ctx context.Context, req *pb.ListNodeLo
CreatedAt: int64(log.CreatedAt),
Count: types.Int32(log.Count),
IsFixed: log.IsFixed == 1,
IsRead: log.IsRead == 1,
})
}
return &pb.ListNodeLogsResponse{NodeLogs: result}, nil
@@ -105,3 +106,33 @@ func (this *NodeLogService) FixNodeLog(ctx context.Context, req *pb.FixNodeLogRe
return this.Success()
}
// CountAllUnreadNodeLogs 计算未读的日志数量
func (this *NodeLogService) CountAllUnreadNodeLogs(ctx context.Context, req *pb.CountAllUnreadNodeLogsRequest) (*pb.RPCCountResponse, error) {
_, err := this.ValidateAdmin(ctx, 0)
if err != nil {
return nil, err
}
var tx = this.NullTx()
count, err := models.SharedNodeLogDAO.CountAllUnreadNodeLogs(tx)
if err != nil {
return nil, err
}
return this.SuccessCount(count)
}
// UpdateNodeLogsRead 设置日志为已读
func (this *NodeLogService) UpdateNodeLogsRead(ctx context.Context, req *pb.UpdateNodeLogsReadRequest) (*pb.RPCSuccess, error) {
_, err := this.ValidateAdmin(ctx, 0)
if err != nil {
return nil, err
}
var tx = this.NullTx()
err = models.SharedNodeLogDAO.UpdateNodeLogsRead(tx, req.NodeLogIds)
if err != nil {
return nil, err
}
return this.Success()
}

View File

@@ -38,11 +38,11 @@ func (this *CommandRequestWaiting) Close() {
close(this.Chan)
}
var responseChanMap = map[int64]*CommandRequestWaiting{} // request id => response
var nodeResponseChanMap = map[int64]*CommandRequestWaiting{} // request id => response
var commandRequestId = int64(0)
var nodeLocker = &sync.Mutex{}
var requestChanMap = map[int64]chan *CommandRequest{} // node id => chan
var nodeRequestChanMap = map[int64]chan *CommandRequest{} // node id => chan
func NextCommandRequestId() int64 {
return atomic.AddInt64(&commandRequestId, 1)
@@ -54,10 +54,10 @@ func init() {
go func() {
for range ticker.C {
nodeLocker.Lock()
for requestId, request := range responseChanMap {
for requestId, request := range nodeResponseChanMap {
if time.Now().Unix()-request.Timestamp > 3600 {
responseChanMap[requestId].Close()
delete(responseChanMap, requestId)
nodeResponseChanMap[requestId].Close()
delete(nodeResponseChanMap, requestId)
}
}
nodeLocker.Unlock()
@@ -127,16 +127,16 @@ func (this *NodeService) NodeStream(server pb.NodeService_NodeStreamServer) erro
}
nodeLocker.Lock()
requestChan, ok := requestChanMap[nodeId]
requestChan, ok := nodeRequestChanMap[nodeId]
if !ok {
requestChan = make(chan *CommandRequest, 1024)
requestChanMap[nodeId] = requestChan
nodeRequestChanMap[nodeId] = requestChan
}
nodeLocker.Unlock()
defer func() {
nodeLocker.Lock()
delete(requestChanMap, nodeId)
delete(nodeRequestChanMap, nodeId)
nodeLocker.Unlock()
}()
@@ -189,7 +189,7 @@ func (this *NodeService) NodeStream(server pb.NodeService_NodeStreamServer) erro
}()
nodeLocker.Lock()
responseChan, ok := responseChanMap[req.RequestId]
responseChan, ok := nodeResponseChanMap[req.RequestId]
if ok {
select {
case responseChan.Chan <- req:
@@ -215,25 +215,37 @@ func (this *NodeService) SendCommandToNode(ctx context.Context, req *pb.NodeStre
return nil, errors.New("node id should not be less than 0")
}
return SendCommandToNode(req.NodeId, req.RequestId, req.Code, req.DataJSON, req.TimeoutSeconds, true)
}
// SendCommandToNode 向节点发送命令
func SendCommandToNode(nodeId int64, requestId int64, messageCode string, dataJSON []byte, timeoutSeconds int32, forceConnecting bool) (result *pb.NodeStreamMessage, err error) {
nodeLocker.Lock()
requestChan, ok := requestChanMap[nodeId]
requestChan, ok := nodeRequestChanMap[nodeId]
nodeLocker.Unlock()
if !ok {
return &pb.NodeStreamMessage{
RequestId: req.RequestId,
IsOk: false,
Message: "node '" + strconv.FormatInt(nodeId, 10) + "' not connected yet",
}, nil
if forceConnecting {
return &pb.NodeStreamMessage{
RequestId: requestId,
IsOk: false,
Message: "node '" + strconv.FormatInt(nodeId, 10) + "' not connected yet",
}, nil
} else {
return &pb.NodeStreamMessage{
RequestId: requestId,
IsOk: true,
}, nil
}
}
req.RequestId = NextCommandRequestId()
requestId = NextCommandRequestId()
select {
case requestChan <- &CommandRequest{
Id: req.RequestId,
Code: req.Code,
CommandJSON: req.DataJSON,
Id: requestId,
Code: messageCode,
CommandJSON: dataJSON,
}:
// 加入到等待队列中
respChan := make(chan *pb.NodeStreamMessage, 1)
@@ -243,11 +255,10 @@ func (this *NodeService) SendCommandToNode(ctx context.Context, req *pb.NodeStre
}
nodeLocker.Lock()
responseChanMap[req.RequestId] = waiting
nodeResponseChanMap[requestId] = waiting
nodeLocker.Unlock()
// 等待响应
timeoutSeconds := req.TimeoutSeconds
if timeoutSeconds <= 0 {
timeoutSeconds = 10
}
@@ -256,14 +267,14 @@ func (this *NodeService) SendCommandToNode(ctx context.Context, req *pb.NodeStre
case resp := <-respChan:
// 从队列中删除
nodeLocker.Lock()
delete(responseChanMap, req.RequestId)
delete(nodeResponseChanMap, requestId)
waiting.Close()
nodeLocker.Unlock()
if resp == nil {
return &pb.NodeStreamMessage{
RequestId: req.RequestId,
Code: req.Code,
RequestId: requestId,
Code: messageCode,
Message: "response timeout",
IsOk: false,
}, nil
@@ -273,21 +284,21 @@ func (this *NodeService) SendCommandToNode(ctx context.Context, req *pb.NodeStre
case <-timeout.C:
// 从队列中删除
nodeLocker.Lock()
delete(responseChanMap, req.RequestId)
delete(nodeResponseChanMap, requestId)
waiting.Close()
nodeLocker.Unlock()
return &pb.NodeStreamMessage{
RequestId: req.RequestId,
Code: req.Code,
RequestId: requestId,
Code: messageCode,
Message: "response timeout over " + fmt.Sprintf("%d", timeoutSeconds) + " seconds",
IsOk: false,
}, nil
}
default:
return &pb.NodeStreamMessage{
RequestId: req.RequestId,
Code: req.Code,
RequestId: requestId,
Code: messageCode,
Message: "command queue is full over " + strconv.Itoa(len(requestChan)),
IsOk: false,
}, nil

View File

@@ -0,0 +1,50 @@
// Copyright 2021 Liuxiangchao iwind.liu@gmail.com. All rights reserved.
//go:build community
// +build community
package services
import (
"context"
"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
)
// PlanService 套餐相关服务
type PlanService struct {
BaseService
}
// CreatePlan 创建套餐
func (this *PlanService) CreatePlan(ctx context.Context, req *pb.CreatePlanRequest) (*pb.CreatePlanResponse, error) {
return &pb.CreatePlanResponse{}, nil
}
// UpdatePlan 修改套餐
func (this *PlanService) UpdatePlan(ctx context.Context, req *pb.UpdatePlanRequest) (*pb.RPCSuccess, error) {
return this.Success()
}
// DeletePlan 删除套餐
func (this *PlanService) DeletePlan(ctx context.Context, req *pb.DeletePlanRequest) (*pb.RPCSuccess, error) {
return this.Success()
}
// FindEnabledPlan 查找单个套餐
func (this *PlanService) FindEnabledPlan(ctx context.Context, req *pb.FindEnabledPlanRequest) (*pb.FindEnabledPlanResponse, error) {
return &pb.FindEnabledPlanResponse{Plan: nil}, nil
}
// CountAllEnabledPlans 计算套餐数量
func (this *PlanService) CountAllEnabledPlans(ctx context.Context, req *pb.CountAllEnabledPlansRequest) (*pb.RPCCountResponse, error) {
return this.SuccessCount(0)
}
// ListEnabledPlans 列出单页套餐
func (this *PlanService) ListEnabledPlans(ctx context.Context, req *pb.ListEnabledPlansRequest) (*pb.ListEnabledPlansResponse, error) {
return &pb.ListEnabledPlansResponse{Plans: nil}, nil
}
// SortPlans 对套餐进行排序
func (this *PlanService) SortPlans(ctx context.Context, req *pb.SortPlansRequest) (*pb.RPCSuccess, error) {
return this.Success()
}

View File

@@ -216,7 +216,7 @@ func (this *ReverseProxyService) UpdateReverseProxy(ctx context.Context, req *pb
}
}
err = models.SharedReverseProxyDAO.UpdateReverseProxy(tx, req.ReverseProxyId, types.Int8(req.RequestHostType), req.RequestHost, req.RequestURI, req.StripPrefix, req.AutoFlush, req.AddHeaders, connTimeout, readTimeout, idleTimeout, req.MaxConns, req.MaxIdleConns)
err = models.SharedReverseProxyDAO.UpdateReverseProxy(tx, req.ReverseProxyId, types.Int8(req.RequestHostType), req.RequestHost, req.RequestURI, req.StripPrefix, req.AutoFlush, req.AddHeaders, connTimeout, readTimeout, idleTimeout, req.MaxConns, req.MaxIdleConns, req.ProxyProtocolJSON)
if err != nil {
return nil, err
}

View File

@@ -8,6 +8,7 @@ import (
"github.com/TeaOSLab/EdgeAPI/internal/db/models"
"github.com/TeaOSLab/EdgeAPI/internal/db/models/dns"
"github.com/TeaOSLab/EdgeAPI/internal/db/models/regions"
"github.com/TeaOSLab/EdgeCommon/pkg/messageconfigs"
"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs"
"github.com/iwind/TeaGo/maps"
@@ -468,6 +469,37 @@ func (this *ServerService) UpdateServerNamesAuditing(ctx context.Context, req *p
return this.Success()
}
// UpdateServerDNS 修改服务的DNS相关设置
func (this *ServerService) UpdateServerDNS(ctx context.Context, req *pb.UpdateServerDNSRequest) (*pb.RPCSuccess, error) {
_, err := this.ValidateAdmin(ctx, 0)
if err != nil {
return nil, err
}
var tx = this.NullTx()
err = models.SharedServerDAO.UpdateServerDNS(tx, req.ServerId, req.SupportCNAME)
if err != nil {
return nil, err
}
return this.Success()
}
// RegenerateServerCNAME 重新生成CNAME
func (this *ServerService) RegenerateServerCNAME(ctx context.Context, req *pb.RegenerateServerCNAMERequest) (*pb.RPCSuccess, error) {
_, err := this.ValidateAdmin(ctx, 0)
if err != nil {
return nil, err
}
var tx = this.NullTx()
_, err = models.SharedServerDAO.GenerateServerDNSName(tx, req.ServerId)
if err != nil {
return nil, err
}
return this.Success()
}
// CountAllEnabledServersMatch 计算服务数量
func (this *ServerService) CountAllEnabledServersMatch(ctx context.Context, req *pb.CountAllEnabledServersMatchRequest) (*pb.RPCCountResponse, error) {
// 校验请求
@@ -712,6 +744,7 @@ func (this *ServerService) FindEnabledServer(ctx context.Context, req *pb.FindEn
Name: server.Name,
Description: server.Description,
DnsName: server.DnsName,
SupportCNAME: server.SupportCNAME == 1,
Config: configJSON,
ServerNamesJSON: []byte(server.ServerNames),
HttpJSON: []byte(server.Http),
@@ -1065,6 +1098,11 @@ func (this *ServerService) FindEnabledServerDNS(ctx context.Context, req *pb.Fin
return nil, err
}
supportCNAME, err := models.SharedServerDAO.FindServerSupportCNAME(tx, req.ServerId)
if err != nil {
return nil, err
}
clusterId, err := models.SharedServerDAO.FindServerClusterId(tx, req.ServerId)
if err != nil {
return nil, err
@@ -1093,8 +1131,9 @@ func (this *ServerService) FindEnabledServerDNS(ctx context.Context, req *pb.Fin
}
return &pb.FindEnabledServerDNSResponse{
DnsName: dnsName,
Domain: pbDomain,
DnsName: dnsName,
Domain: pbDomain,
SupportCNAME: supportCNAME,
}, nil
}
@@ -1521,3 +1560,137 @@ func (this *ServerService) FindNearbyServers(ctx context.Context, req *pb.FindNe
Groups: []*pb.FindNearbyServersResponse_GroupInfo{pbGroup},
}, nil
}
// PurgeServerCache 清除缓存
func (this *ServerService) PurgeServerCache(ctx context.Context, req *pb.PurgeServerCacheRequest) (*pb.PurgeServerCacheResponse, error) {
_, err := this.ValidateAdmin(ctx, 0)
if err != nil {
// 检查是否为节点
_, err = this.ValidateNode(ctx)
if err != nil {
return nil, err
}
}
if len(req.Domains) == 0 {
return nil, errors.New("'domains' field is required")
}
if len(req.Keys) == 0 && len(req.Prefixes) == 0 {
return &pb.PurgeServerCacheResponse{IsOk: true}, nil
}
var tx = this.NullTx()
var cacheMap = maps.Map{}
var purgeResponse = &pb.PurgeServerCacheResponse{}
for _, domain := range req.Domains {
servers, err := models.SharedServerDAO.FindAllEnabledServersWithDomain(tx, domain)
if err != nil {
return nil, err
}
for _, server := range servers {
clusterId := int64(server.ClusterId)
if clusterId > 0 {
nodeIds, err := models.SharedNodeDAO.FindAllEnabledNodeIdsWithClusterId(tx, clusterId)
if err != nil {
return nil, err
}
cachePolicyId, err := models.SharedNodeClusterDAO.FindClusterHTTPCachePolicyId(tx, clusterId, cacheMap)
if err != nil {
return nil, err
}
if cachePolicyId == 0 {
continue
}
cachePolicy, err := models.SharedHTTPCachePolicyDAO.ComposeCachePolicy(tx, cachePolicyId, cacheMap)
if err != nil {
return nil, err
}
if cachePolicy == nil {
continue
}
cachePolicyJSON, err := json.Marshal(cachePolicy)
if err != nil {
return nil, err
}
for _, nodeId := range nodeIds {
msg := &messageconfigs.PurgeCacheMessage{
CachePolicyJSON: cachePolicyJSON,
}
if len(req.Prefixes) > 0 {
msg.Type = messageconfigs.PurgeCacheMessageTypeDir
msg.Keys = req.Prefixes
} else {
msg.Type = messageconfigs.PurgeCacheMessageTypeFile
msg.Keys = req.Keys
}
msgJSON, err := json.Marshal(msg)
if err != nil {
return nil, err
}
resp, err := SendCommandToNode(nodeId, NextCommandRequestId(), messageconfigs.MessageCodePurgeCache, msgJSON, 10, false)
if err != nil {
return nil, err
}
if !resp.IsOk {
purgeResponse.IsOk = false
purgeResponse.Message = resp.Message
return purgeResponse, nil
}
}
}
}
}
purgeResponse.IsOk = true
return purgeResponse, nil
}
// FindEnabledServerBandwidthLimit 查找带宽限制
func (this *ServerService) FindEnabledServerBandwidthLimit(ctx context.Context, req *pb.FindEnabledServerBandwidthLimitRequest) (*pb.FindEnabledServerBandwidthLimitResponse, error) {
_, _, err := this.ValidateAdminAndUser(ctx, 0, 0)
if err != nil {
return nil, err
}
var tx = this.NullTx()
limitConfig, err := models.SharedServerDAO.FindServerBandwidthLimitConfig(tx, req.ServerId, nil)
if err != nil {
return nil, err
}
limitConfigJSON, err := json.Marshal(limitConfig)
if err != nil {
return nil, err
}
return &pb.FindEnabledServerBandwidthLimitResponse{
BandwidthLimitJSON: limitConfigJSON,
}, nil
}
// UpdateServerBandwidthLimit 设置带宽限制
func (this *ServerService) UpdateServerBandwidthLimit(ctx context.Context, req *pb.UpdateServerBandwidthLimitRequest) (*pb.RPCSuccess, error) {
_, _, err := this.ValidateAdminAndUser(ctx, 0, 0)
if err != nil {
return nil, err
}
var tx = this.NullTx()
var config = &serverconfigs.BandwidthLimitConfig{}
err = json.Unmarshal(req.BandwidthLimitJSON, config)
if err != nil {
return nil, err
}
err = models.SharedServerDAO.UpdateServerBandwidthLimitConfig(tx, req.ServerId, config)
if err != nil {
return nil, err
}
return this.Success()
}

View File

@@ -25,6 +25,7 @@ func (this *ServerDailyStatService) UploadServerDailyStats(ctx context.Context,
tx := this.NullTx()
// 保存统计数据
err = models.SharedServerDailyStatDAO.SaveStats(tx, req.Stats)
if err != nil {
return nil, err

View File

@@ -0,0 +1,45 @@
// Copyright 2021 Liuxiangchao iwind.liu@gmail.com. All rights reserved.
//go:build community
// +build community
package services
import (
"context"
"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
)
// UserPlanService 用户购买的套餐
type UserPlanService struct {
BaseService
}
// CreateUserPlan 添加已购套餐
func (this *UserPlanService) CreateUserPlan(ctx context.Context, req *pb.CreateUserPlanRequest) (*pb.CreateUserPlanResponse, error) {
return &pb.CreateUserPlanResponse{UserPlanId: 0}, nil
}
// FindEnabledUserPlan 查找单个已购套餐信息
func (this *UserPlanService) FindEnabledUserPlan(ctx context.Context, req *pb.FindEnabledUserPlanRequest) (*pb.FindEnabledUserPlanResponse, error) {
return &pb.FindEnabledUserPlanResponse{UserPlan: nil}, nil
}
// UpdateUserPlan 修改已购套餐
func (this *UserPlanService) UpdateUserPlan(ctx context.Context, req *pb.UpdateUserPlanRequest) (*pb.RPCSuccess, error) {
return this.Success()
}
// DeleteUserPlan 删除已购套餐
func (this *UserPlanService) DeleteUserPlan(ctx context.Context, req *pb.DeleteUserPlanRequest) (*pb.RPCSuccess, error) {
return this.Success()
}
// CountAllEnabledUserPlans 计算已购套餐数
func (this *UserPlanService) CountAllEnabledUserPlans(ctx context.Context, req *pb.CountAllEnabledUserPlansRequest) (*pb.RPCCountResponse, error) {
return this.SuccessCount(0)
}
// ListEnabledUserPlans 列出单页已购套餐
func (this *UserPlanService) ListEnabledUserPlans(ctx context.Context, req *pb.ListEnabledUserPlansRequest) (*pb.ListEnabledUserPlansResponse, error) {
return &pb.ListEnabledUserPlansResponse{UserPlans: nil}, nil
}

File diff suppressed because one or more lines are too long

View File

@@ -56,6 +56,9 @@ var upgradeFuncs = []*upgradeVersion{
{
"0.3.2", upgradeV0_3_2,
},
{
"0.3.3", upgradeV0_3_3,
},
}
// UpgradeSQLData 升级SQL数据
@@ -408,7 +411,7 @@ func upgradeV0_3_2(db *dbs.DB) error {
GzipId int64 `yaml:"gzipId" json:"gzipId"` // 使用的配置ID
}
webOnes, _, err := db.FindOnes("SELECT id, gzip FROM edgeHTTPWebs WHERE gzip IS NOT NULL")
webOnes, _, err := db.FindOnes("SELECT id, gzip FROM edgeHTTPWebs WHERE gzip IS NOT NULL AND compression IS NULL")
if err != nil {
return err
}
@@ -511,3 +514,14 @@ func upgradeV0_3_2(db *dbs.DB) error {
return nil
}
// v0.3.3
func upgradeV0_3_3(db *dbs.DB) error {
// 升级CC请求数Code
_, err := db.Exec("UPDATE edgeHTTPFirewallRuleSets SET code='8002' WHERE name='CC请求数' AND code='8001'")
if err != nil {
return err
}
return nil
}

View File

@@ -22,7 +22,7 @@ func TestUpgradeSQLData(t *testing.T) {
}
func TestUpgradeSQLData_v1_3_1(t *testing.T) {
func TestUpgradeSQLData_v0_3_1(t *testing.T) {
db, err := dbs.NewInstanceFromConfig(&dbs.DBConfig{
Driver: "mysql",
Dsn: "root:123456@tcp(127.0.0.1:3306)/db_edge_new?charset=utf8mb4&timeout=30s",
@@ -38,7 +38,7 @@ func TestUpgradeSQLData_v1_3_1(t *testing.T) {
t.Log("ok")
}
func TestUpgradeSQLData_v1_3_2(t *testing.T) {
func TestUpgradeSQLData_v0_3_2(t *testing.T) {
db, err := dbs.NewInstanceFromConfig(&dbs.DBConfig{
Driver: "mysql",
Dsn: "root:123456@tcp(127.0.0.1:3306)/db_edge?charset=utf8mb4&timeout=30s",
@@ -52,4 +52,20 @@ func TestUpgradeSQLData_v1_3_2(t *testing.T) {
t.Fatal(err)
}
t.Log("ok")
}
func TestUpgradeSQLData_v0_3_3(t *testing.T) {
db, err := dbs.NewInstanceFromConfig(&dbs.DBConfig{
Driver: "mysql",
Dsn: "root:123456@tcp(127.0.0.1:3306)/db_edge?charset=utf8mb4&timeout=30s",
Prefix: "edge",
})
if err != nil {
t.Fatal(err)
}
err = upgradeV0_3_3(db)
if err != nil {
t.Fatal(err)
}
t.Log("ok")
}

View File

@@ -10,8 +10,10 @@ import (
"github.com/TeaOSLab/EdgeAPI/internal/utils"
"github.com/TeaOSLab/EdgeCommon/pkg/configutils"
"github.com/TeaOSLab/EdgeCommon/pkg/nodeconfigs"
"github.com/TeaOSLab/EdgeCommon/pkg/nodeutils"
"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs"
"github.com/iwind/TeaGo/lists"
"github.com/iwind/TeaGo/maps"
"github.com/iwind/TeaGo/types"
"net"
"net/http"
@@ -167,11 +169,28 @@ func (this *HealthCheckExecutor) checkNode(healthCheckConfig *serverconfigs.Heal
result.NodeAddr = "[" + result.NodeAddr + "]"
}
if len(healthCheckConfig.URL) == 0 {
healthCheckConfig.URL = "http://${host}/"
}
url := strings.ReplaceAll(healthCheckConfig.URL, "${host}", result.NodeAddr)
req, err := http.NewRequest(http.MethodGet, url, nil)
if err != nil {
return err
}
if len(healthCheckConfig.UserAgent) > 0 {
req.Header.Set("User-Agent", healthCheckConfig.UserAgent)
} else {
req.Header.Set("User-Agent", "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/94.0.4606.81 Safari/537.36")
}
key, err := nodeutils.EncryptData(result.Node.UniqueId, result.Node.Secret, maps.Map{
"onlyBasicRequest": healthCheckConfig.OnlyBasicRequest,
}, 300)
if err != nil {
return err
}
req.Header.Set(serverconfigs.HealthCheckHeaderName, key)
timeout := 5 * time.Second
if healthCheckConfig.Timeout != nil {