432 lines
11 KiB
Go
432 lines
11 KiB
Go
package service
|
||
|
||
import (
|
||
"context"
|
||
"fmt"
|
||
"sort"
|
||
"time"
|
||
|
||
"go-common/app/admin/main/aegis/model"
|
||
"go-common/app/admin/main/aegis/model/net"
|
||
"go-common/library/ecode"
|
||
"go-common/library/log"
|
||
|
||
"github.com/jinzhu/gorm"
|
||
)
|
||
|
||
/**
|
||
* 业务下所有需要分发任务的flow,用net+flow中文名拼接
|
||
*/
|
||
func (s *Service) dispatchFlow(c context.Context, businessID []int64, limitFlow []int64) (res map[int64]map[int64]string, err error) {
|
||
var (
|
||
nets []*net.Net
|
||
tranID []int64
|
||
dirs []*net.Direction
|
||
flows []*net.Flow
|
||
)
|
||
|
||
//biz:flow:flow_name
|
||
res = map[int64]map[int64]string{}
|
||
//业务下所有可用网
|
||
if nets, err = s.gorm.NetsByBusiness(c, businessID, false); err != nil {
|
||
log.Error("dispatchFlow s.gorm.NetsByBusiness(%v) error(%v)", businessID, err)
|
||
return
|
||
}
|
||
netID := []int64{}
|
||
netMap := map[int64]*net.Net{}
|
||
for _, item := range nets {
|
||
netID = append(netID, item.ID)
|
||
netMap[item.ID] = item
|
||
}
|
||
//网下所有可用变迁
|
||
if tranID, err = s.tranIDByNet(c, netID, true, false); err != nil {
|
||
log.Error("dispatchFlow s.gorm.TransitionIDByNet(%v) error(%v) businessid(%d)", netID, err, businessID)
|
||
return
|
||
}
|
||
if len(tranID) == 0 {
|
||
return
|
||
}
|
||
//变迁所有可用的被指向的有向线
|
||
if dirs, err = s.gorm.DirectionByTransitionID(c, tranID, net.DirInput, false); err != nil {
|
||
log.Error("dispatchFlow s.gorm.DirectionByTransitionID error(%v) businessid(%d)", err, businessID)
|
||
return
|
||
}
|
||
limitFlowMap := map[int64]int{}
|
||
for _, item := range limitFlow {
|
||
limitFlowMap[item] = 1
|
||
}
|
||
accessFlow := []int64{}
|
||
for _, item := range dirs {
|
||
if len(limitFlowMap) > 0 && limitFlowMap[item.FlowID] <= 0 {
|
||
continue
|
||
}
|
||
accessFlow = append(accessFlow, item.FlowID)
|
||
}
|
||
if len(accessFlow) == 0 {
|
||
return
|
||
}
|
||
//拼接每个节点的中文名
|
||
if flows, err = s.flows(c, accessFlow, false); err != nil {
|
||
log.Error("dispatchFlow s.flows error(%v) businessid(%d)", err, businessID)
|
||
return
|
||
}
|
||
sort.Sort(net.FlowArr(flows))
|
||
for _, item := range flows {
|
||
if item == nil || netMap[item.NetID] == nil {
|
||
continue
|
||
}
|
||
|
||
nt := netMap[item.NetID]
|
||
if _, exist := res[nt.BusinessID]; !exist {
|
||
res[nt.BusinessID] = map[int64]string{}
|
||
}
|
||
res[nt.BusinessID][item.ID] = nt.ChName + item.ChName
|
||
}
|
||
return
|
||
}
|
||
|
||
//ShowFlow .
|
||
func (s *Service) ShowFlow(c context.Context, id int64) (r *net.ShowFlowResult, err error) {
|
||
var (
|
||
f *net.Flow
|
||
details map[int64][]*net.TokenBind
|
||
n *net.Net
|
||
)
|
||
|
||
if f, err = s.gorm.FlowByID(c, id); err != nil {
|
||
return
|
||
}
|
||
if details, err = s.gorm.TokenBindByElement(c, []int64{id}, []int8{net.BindTypeFlow}, true); err != nil {
|
||
return
|
||
}
|
||
if n, err = s.gorm.NetByID(c, f.NetID); err != nil {
|
||
return
|
||
}
|
||
r = &net.ShowFlowResult{
|
||
Flow: f,
|
||
Tokens: details[id],
|
||
IsStart: n.StartFlowID == id,
|
||
}
|
||
return
|
||
}
|
||
|
||
//GetFlowList .
|
||
func (s *Service) GetFlowList(c context.Context, pm *net.ListNetElementParam) (result *net.ListFlowRes, err error) {
|
||
var (
|
||
flowID []int64
|
||
tks map[int64][]*net.TokenBind
|
||
n *net.Net
|
||
uid = []int64{}
|
||
unames map[int64]string
|
||
)
|
||
|
||
if result, err = s.gorm.FlowList(c, pm); err != nil {
|
||
return
|
||
}
|
||
if len(result.Result) == 0 {
|
||
return
|
||
}
|
||
for _, item := range result.Result {
|
||
flowID = append(flowID, item.ID)
|
||
uid = append(uid, item.UID)
|
||
}
|
||
if tks, err = s.gorm.TokenBindByElement(c, flowID, []int8{net.BindTypeFlow}, true); err != nil {
|
||
return
|
||
}
|
||
if n, err = s.gorm.NetByID(c, pm.NetID); err != nil {
|
||
return
|
||
}
|
||
if unames, err = s.http.GetUnames(c, uid); err != nil {
|
||
log.Error("GetFlowList s.http.GetUnames error(%v)", err)
|
||
err = nil
|
||
}
|
||
for _, item := range result.Result {
|
||
item.IsStart = item.ID == n.StartFlowID
|
||
item.Username = unames[item.UID]
|
||
for _, bd := range tks[item.ID] {
|
||
item.Tokens = append(item.Tokens, bd.ChName)
|
||
}
|
||
}
|
||
|
||
return
|
||
}
|
||
|
||
//GetFlowByNet .
|
||
func (s *Service) GetFlowByNet(c context.Context, netID int64) (result map[int64]string, err error) {
|
||
var (
|
||
flows []*net.Flow
|
||
)
|
||
result = map[int64]string{}
|
||
if flows, err = s.gorm.FlowsByNet(c, []int64{netID}); err != nil {
|
||
log.Error("GetFlowByNet s.gorm.FlowsByNet(%d) error(%v)", netID, err)
|
||
return
|
||
}
|
||
|
||
for _, item := range flows {
|
||
result[item.ID] = item.ChName
|
||
}
|
||
return
|
||
}
|
||
|
||
func (s *Service) checkFlowUnique(c context.Context, netID int64, name string) (err error, msg string) {
|
||
var exist *net.Flow
|
||
if exist, err = s.gorm.FlowByUnique(c, netID, name); err != nil {
|
||
log.Error("checkFlowUnique s.gorm.FlowByUnique(%d,%s) error(%v)", netID, name, err)
|
||
return
|
||
}
|
||
if exist != nil {
|
||
err = ecode.AegisUniqueAlreadyExist
|
||
msg = fmt.Sprintf(ecode.AegisUniqueAlreadyExist.Message(), "节点", name)
|
||
}
|
||
return
|
||
}
|
||
|
||
func (s *Service) checkStartFlowBind(oldFlow *net.Flow, tokenIDList []int64) (err error, msg string) {
|
||
if oldFlow != nil && !oldFlow.IsAvailable() {
|
||
err = ecode.AegisFlowDisabled
|
||
msg = fmt.Sprintf("%s,不能作为初始节点", ecode.AegisFlowDisabled.Message())
|
||
return
|
||
}
|
||
//第一版动态审核初始接入状态：敏感待审、非敏感待审、高频转发待审，非一个确定性值，而系统不提供条件判断和guard解析，由
|
||
//配置初始节点没有令牌，而业务start时自动传入state支持(后续接入统一初始状态进而条件分状态的逻辑后，去掉state字段，且加上该判断)
|
||
//if len(tokenIDList) == 0 {
|
||
// err = ecode.AegisFlowNoToken
|
||
// msg = fmt.Sprintf("%s,不能作为初始节点", ecode.AegisFlowNoToken.Message())
|
||
//}
|
||
|
||
return
|
||
}
|
||
|
||
//AddFlow .
|
||
func (s *Service) AddFlow(c context.Context, uid int64, f *net.FlowEditParam) (id int64, err error, msg string) {
|
||
var (
|
||
tx *gorm.DB
|
||
diff = []string{}
|
||
diffBind string
|
||
)
|
||
if err, msg = s.checkFlowUnique(c, f.NetID, f.Name); err != nil {
|
||
return
|
||
}
|
||
if f.IsStart {
|
||
if err, msg = s.checkStartFlowBind(nil, f.TokenIDList); err != nil {
|
||
return
|
||
}
|
||
diff = append(diff, model.LogFieldTemp(model.LogFieldStartFlow, f.IsStart, false, false))
|
||
}
|
||
flow := &net.Flow{
|
||
NetID: f.NetID,
|
||
Name: f.Name,
|
||
ChName: f.ChName,
|
||
Description: f.Description,
|
||
UID: uid,
|
||
}
|
||
|
||
//db update
|
||
tx, err = s.gorm.BeginTx(c)
|
||
if err != nil {
|
||
log.Error("AddFlow s.gorm.BeginTx error(%v)", err)
|
||
return
|
||
}
|
||
if err = s.gorm.AddItem(c, tx, flow); err != nil {
|
||
tx.Rollback()
|
||
return
|
||
}
|
||
if diffBind, _, err, msg = s.compareFlowBind(c, tx, flow.ID, f.TokenIDList, false); err != nil {
|
||
log.Error("AddFlow s.compareFlowBind error(%v) params(%+v)", err, f)
|
||
tx.Rollback()
|
||
return
|
||
}
|
||
if diffBind != "" {
|
||
diff = append(diff, diffBind)
|
||
}
|
||
if f.IsStart {
|
||
if err = s.gorm.NetBindStartFlow(c, tx, flow.NetID, flow.ID); err != nil {
|
||
tx.Rollback()
|
||
return
|
||
}
|
||
}
|
||
if err = tx.Commit().Error; err != nil {
|
||
log.Error("AddFlow tx.Commit error(%v)", err)
|
||
return
|
||
}
|
||
|
||
id = flow.ID
|
||
|
||
//日志
|
||
diff = append(diff, model.LogFieldTemp(model.LogFieldChName, f.ChName, "", false))
|
||
diff = append(diff, model.LogFieldTemp(model.LogFieldName, f.Name, "", false))
|
||
oper := &model.NetConfOper{
|
||
OID: flow.ID,
|
||
Action: model.LogNetActionNew,
|
||
UID: flow.UID,
|
||
NetID: flow.NetID,
|
||
ChName: flow.ChName,
|
||
FlowID: flow.ID,
|
||
Diff: diff,
|
||
}
|
||
s.sendNetConfLog(c, model.LogTypeFlowConf, oper)
|
||
return
|
||
}
|
||
|
||
// UpdateFlow .
|
||
func (s *Service) UpdateFlow(c context.Context, uid int64, f *net.FlowEditParam) (err error, msg string) {
|
||
var (
|
||
old *net.Flow
|
||
n *net.Net
|
||
startFlowID int64 = -1
|
||
updates = map[string]interface{}{}
|
||
tx *gorm.DB
|
||
diff = []string{}
|
||
diffBind string
|
||
changedBind []int64
|
||
)
|
||
if old, err = s.gorm.FlowByID(c, f.ID); err != nil {
|
||
log.Error("UpdateFlow s.gorm.FlowByID(%d) error(%v)", f.ID, err)
|
||
return
|
||
}
|
||
if n, err = s.gorm.NetByID(c, old.NetID); err != nil {
|
||
log.Error("UpdateFlow s.gorm.NetByID(%d) error(%v) flowid(%d)", old.NetID, err, f.ID)
|
||
return
|
||
}
|
||
if f.IsStart && n.StartFlowID != f.ID {
|
||
startFlowID = f.ID
|
||
} else if !f.IsStart && n.StartFlowID == f.ID {
|
||
startFlowID = 0
|
||
}
|
||
|
||
if f.IsStart {
|
||
if err, msg = s.checkStartFlowBind(old, f.TokenIDList); err != nil {
|
||
return
|
||
}
|
||
diff = append(diff, model.LogFieldTemp(model.LogFieldStartFlow, true, false, true))
|
||
}
|
||
|
||
if f.Name != old.Name {
|
||
if err, msg = s.checkFlowUnique(c, old.NetID, f.Name); err != nil {
|
||
return
|
||
}
|
||
diff = append(diff, model.LogFieldTemp(model.LogFieldName, f.Name, old.Name, true))
|
||
old.Name = f.Name
|
||
updates["name"] = f.Name
|
||
}
|
||
if f.ChName != old.ChName {
|
||
diff = append(diff, model.LogFieldTemp(model.LogFieldChName, f.ChName, old.ChName, true))
|
||
old.ChName = f.ChName
|
||
updates["ch_name"] = f.ChName
|
||
}
|
||
if f.Description != old.Description {
|
||
old.Description = f.Description
|
||
updates["description"] = f.Description
|
||
}
|
||
|
||
//db update
|
||
tx, err = s.gorm.BeginTx(c)
|
||
if err != nil {
|
||
log.Error("UpdateFlow s.gorm.BeginTx error(%v)", err)
|
||
return
|
||
}
|
||
if len(updates) > 0 {
|
||
if err = s.gorm.UpdateFields(c, tx, net.TableFlow, old.ID, updates); err != nil {
|
||
tx.Rollback()
|
||
return
|
||
}
|
||
}
|
||
if startFlowID >= 0 {
|
||
if err = s.gorm.NetBindStartFlow(c, tx, n.ID, startFlowID); err != nil {
|
||
tx.Rollback()
|
||
return
|
||
}
|
||
}
|
||
if diffBind, changedBind, err, msg = s.compareFlowBind(c, tx, f.ID, f.TokenIDList, true); err != nil {
|
||
log.Error("updateFlow s.compareFlowBind error(%v) params(%+v)", err, f)
|
||
tx.Rollback()
|
||
return
|
||
}
|
||
if diffBind != "" {
|
||
diff = append(diff, diffBind)
|
||
}
|
||
if err = tx.Commit().Error; err != nil {
|
||
log.Error("UpdateFlow tx.Commit error(%v)", err)
|
||
return
|
||
}
|
||
s.delFlowCache(c, old, changedBind)
|
||
|
||
//日志
|
||
if len(diff) == 0 {
|
||
return
|
||
}
|
||
oper := &model.NetConfOper{
|
||
OID: old.ID,
|
||
Action: model.LogNetActionUpdate,
|
||
UID: uid,
|
||
NetID: old.NetID,
|
||
ChName: old.ChName,
|
||
FlowID: old.ID,
|
||
Diff: diff,
|
||
}
|
||
s.sendNetConfLog(c, model.LogTypeFlowConf, oper)
|
||
return
|
||
}
|
||
|
||
//SwitchFlow .
|
||
func (s *Service) SwitchFlow(c context.Context, id int64, needDisable bool) (err error) {
|
||
var (
|
||
old *net.Flow
|
||
n *net.Net
|
||
dirs []*net.Direction
|
||
action string
|
||
)
|
||
if old, err = s.gorm.FlowByID(c, id); err != nil {
|
||
log.Error("SwitchFlow s.gorm.FlowByID(%d) error(%v) needDisable(%v)", id, err, needDisable)
|
||
return
|
||
}
|
||
log.Info("SwitchFlow id(%d) needdisable(%v) old-flow(%+v)", id, needDisable, old)
|
||
available := old.IsAvailable()
|
||
if available == !needDisable {
|
||
return
|
||
}
|
||
|
||
if needDisable {
|
||
if dirs, err = s.gorm.DirectionByFlowID(c, []int64{id}, 0); err != nil {
|
||
log.Error("SwitchFlow s.gorm.DirectionByFlowID(%d) error(%v)", id, err)
|
||
return
|
||
}
|
||
if len(dirs) > 0 {
|
||
log.Error("SwitchFlow dir by flow(%d) founded", id)
|
||
err = ecode.AegisFlowBinded
|
||
return
|
||
}
|
||
if n, err = s.gorm.NetByID(c, old.NetID); err != nil {
|
||
log.Error("SwitchFlow s.gorm.NetByID(%d) error(%v) flow(%d)", old.NetID, err, id)
|
||
return
|
||
}
|
||
if n.StartFlowID == id {
|
||
log.Error("SwitchFlow net(%d).startflow=flow(%d) founded", n.ID, id)
|
||
err = ecode.AegisFlowBinded
|
||
return
|
||
}
|
||
old.DisableTime = time.Now()
|
||
action = model.LogNetActionDisable
|
||
} else {
|
||
old.DisableTime = net.Recovered
|
||
action = model.LogNetActionAvailable
|
||
}
|
||
|
||
if err = s.gorm.UpdateFields(c, nil, net.TableFlow, id, map[string]interface{}{"disable_time": old.DisableTime}); err != nil {
|
||
return
|
||
}
|
||
s.delFlowCache(c, old, nil)
|
||
|
||
//日志
|
||
oper := &model.NetConfOper{
|
||
OID: old.ID,
|
||
Action: action,
|
||
UID: old.UID,
|
||
NetID: old.NetID,
|
||
ChName: old.ChName,
|
||
FlowID: old.ID,
|
||
}
|
||
s.sendNetConfLog(c, model.LogTypeFlowConf, oper)
|
||
return
|
||
}
|