You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
318 lines
8.1 KiB
318 lines
8.1 KiB
package app
|
|
|
|
import (
|
|
"dataSource"
|
|
"encoding/gob"
|
|
"errors"
|
|
"et_prometheus_exporter"
|
|
"fmt"
|
|
"gitea.anxinyun.cn/container/common_models"
|
|
"gitea.anxinyun.cn/container/common_utils/configLoad"
|
|
"log"
|
|
"net"
|
|
"net/rpc"
|
|
"sort"
|
|
"time"
|
|
)
|
|
|
|
func NewEtMaster() EtMaster {
|
|
master := EtMaster{
|
|
exporter: et_prometheus_exporter.NewPrometheusExporter(),
|
|
sleepCH: make(chan bool, 1),
|
|
}
|
|
return master
|
|
}
|
|
|
|
type EtMaster struct {
|
|
nodeList []*NodeRpc
|
|
exporter et_prometheus_exporter.PrometheusExporter
|
|
sleepCH chan bool
|
|
}
|
|
|
|
type NodeRpc struct {
|
|
args *common_models.NodeArgs // 注册节点参数：RPC服务名为 master, 服务方法 NodeRegister 的输入参数
|
|
resultCH chan bool // 注册节点参数：RPC服务名为 master, 服务方法 NodeRegister 的输出结果
|
|
client *rpc.Client
|
|
}
|
|
|
|
// RegisterListen 启动 master RPC服务
|
|
func (the *EtMaster) RegisterListen() {
|
|
//监听
|
|
err := rpc.RegisterName("master", the)
|
|
if err != nil {
|
|
log.Println("master 提供注册服务异常")
|
|
return
|
|
}
|
|
|
|
port := configLoad.LoadConfig().GetUint16("master.port")
|
|
|
|
listener, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
|
|
if err != nil {
|
|
log.Panic("master 启动 node服务注册功能异常")
|
|
}
|
|
log.Printf("master 启动 node服务注册功能 :%d", port)
|
|
for {
|
|
//log.Println("master 监听新注册链接")
|
|
conn, err := listener.Accept()
|
|
if err != nil {
|
|
log.Println("master rpc Accept异常")
|
|
}
|
|
log.Printf("master Accept注册链接 from node[%s]", conn.RemoteAddr())
|
|
go rpc.ServeConn(conn)
|
|
}
|
|
}
|
|
|
|
// NodeRegister 是 RPC 服务方法，由 et_node 远程调用
|
|
func (the *EtMaster) NodeRegister(nodeArgs *common_models.NodeArgs, replay *bool) error {
|
|
node := &NodeRpc{
|
|
args: nodeArgs,
|
|
resultCH: make(chan bool, 1),
|
|
client: nil,
|
|
}
|
|
//master 初始化 node client
|
|
client, err := rpc.Dial("tcp", node.args.Addr)
|
|
if err != nil {
|
|
log.Printf("链接node失败-> node[%v]", node.args.Addr)
|
|
return err
|
|
}
|
|
node.client = client
|
|
|
|
the.nodeList = append(the.nodeList, node)
|
|
log.Printf("node服务[%v] 注册成功", nodeArgs)
|
|
printNodesInfo(the.nodeList)
|
|
*replay = true
|
|
return nil
|
|
}
|
|
func (the *EtMaster) NodeHeart(nodeArgs *common_models.NodeArgs, replay *bool) error {
|
|
|
|
isRegister := false
|
|
for _, nodeRpc := range the.nodeList {
|
|
if nodeRpc.args.Addr == nodeArgs.Addr {
|
|
isRegister = true
|
|
}
|
|
}
|
|
if !isRegister {
|
|
log.Printf("收到-未注册的node[%v] 心跳", nodeArgs)
|
|
*replay = false
|
|
return errors.New("未注册的node")
|
|
}
|
|
|
|
log.Printf("收到-node[%v] 心跳", nodeArgs)
|
|
*replay = true
|
|
|
|
return nil
|
|
}
|
|
|
|
// NodeUnRegister 节点RPC 注销
|
|
func (the *EtMaster) NodeUnRegister(nodeArgs *common_models.NodeArgs, replay *bool) error {
|
|
|
|
for i, node := range the.nodeList {
|
|
log.Printf("节点[%s] 注销", node.args.Addr)
|
|
if node.args.Addr == nodeArgs.Addr {
|
|
err := node.client.Close()
|
|
if err != nil {
|
|
log.Printf("节点[%s] client关闭异常 %s", node.args.Addr, err.Error())
|
|
}
|
|
the.nodeList[i] = nil
|
|
}
|
|
}
|
|
the.nodesTidy()
|
|
log.Printf("node服务[%v] 注销成功", nodeArgs)
|
|
*replay = true
|
|
return nil
|
|
}
|
|
|
|
func (the *EtMaster) nodesTidy() {
|
|
the.nodeList = updateNodeList(the.nodeList)
|
|
printNodesInfo(the.nodeList)
|
|
}
|
|
|
|
func updateNodeList(nodes []*NodeRpc) []*NodeRpc {
|
|
var newNodes []*NodeRpc
|
|
for _, node := range nodes {
|
|
if node != nil && node.client != nil {
|
|
newNodes = append(newNodes, node)
|
|
}
|
|
}
|
|
return newNodes
|
|
}
|
|
func printNodesInfo(nodes []*NodeRpc) {
|
|
info := fmt.Sprintf("共[%d]个节点:\n ", len(nodes))
|
|
for _, node := range nodes {
|
|
info += fmt.Sprintf("%s,%s\n", node.args.ID, node.args.Addr)
|
|
}
|
|
log.Println(info)
|
|
|
|
}
|
|
|
|
func (the *EtMaster) WaitNodeRegister() {
|
|
log.Println("等待 node进行注册")
|
|
for {
|
|
if len(the.nodeList) > 0 {
|
|
break
|
|
}
|
|
time.Sleep(time.Second * 10)
|
|
}
|
|
}
|
|
func (the *EtMaster) ConnectNode() {
|
|
for i := range the.nodeList {
|
|
nodeAddr := the.nodeList[i].args.Addr
|
|
if the.nodeList[i].client == nil {
|
|
client, err := rpc.Dial("tcp", nodeAddr)
|
|
if err != nil {
|
|
log.Printf("链接node失败-> node[%v]", nodeAddr)
|
|
continue
|
|
}
|
|
the.nodeList[i].client = client
|
|
}
|
|
}
|
|
}
|
|
|
|
func (the *EtMaster) call_etNode(node *NodeRpc, args *common_models.IotaData) {
|
|
if node.client == nil {
|
|
log.Printf("node [%v] client=nil", node.args)
|
|
return
|
|
}
|
|
|
|
the.exporter.OnIotaData2metricByPrometheus(args)
|
|
resultCH := make(chan bool, 1)
|
|
go func() {
|
|
defer timeCost(node.args.ID, args.DeviceId, time.Now())
|
|
var rpcResult bool
|
|
err := node.client.Call("etNode.Handler", args, &rpcResult)
|
|
if err != nil {
|
|
log.Printf("rpc 调用node, err:%s", err.Error())
|
|
}
|
|
resultCH <- rpcResult
|
|
}()
|
|
|
|
//等result
|
|
var result bool
|
|
timeOut := 1000 * time.Millisecond
|
|
select {
|
|
case result = <-resultCH:
|
|
case <-time.After(timeOut):
|
|
log.Printf("node[%s]处理[%s|%s]超过 %v,超时", node.args.Addr, args.DeviceId, args.TriggerTime, timeOut)
|
|
result = false
|
|
}
|
|
//log.Printf("node[%s]处理[%s|%s]结果=%v", node.args.Addr, args.DeviceId, args.TriggerTime, result)
|
|
if result == false {
|
|
//发送 stop 信号
|
|
|
|
the.sleepCH <- true
|
|
log.Println("=============================================")
|
|
log.Printf("node[%s]处理[%s|%s]异常,触发nodesTidy", node.args.Addr, args.DeviceId, args.TriggerTime)
|
|
time.Sleep(time.Second * 5)
|
|
node.client = nil
|
|
the.nodesTidy()
|
|
}
|
|
}
|
|
|
|
func (the *EtMaster) call_aggNode(node *NodeRpc, args *common_models.AggData) {
|
|
if node.client == nil {
|
|
log.Printf("et node [%v] client=nil", node.args)
|
|
return
|
|
}
|
|
go func() {
|
|
var response bool
|
|
err := node.client.Call("aggNode.Handler", args, &response)
|
|
if err != nil {
|
|
log.Printf("rpc err:%s", err.Error())
|
|
}
|
|
// 将 rpc 返回值写入 chan
|
|
node.resultCH <- response
|
|
}()
|
|
|
|
// 接收数据
|
|
var result bool
|
|
select {
|
|
case result = <-node.resultCH:
|
|
case <-time.After(2 * time.Second):
|
|
log.Println("rpc [aggHandler.Handler] 调用超时，退出。")
|
|
result = false
|
|
}
|
|
log.Printf("agg node[%s]处理结果:%v", node.args.Addr, result)
|
|
}
|
|
|
|
// DistributeData 分发源数据给各类型处理节点。
|
|
// 通过不断从不同类型的数据通道中读取数据，并根据节点类型分发数据进行处理，可以实现数据的采集和分发功能。
|
|
func (the *EtMaster) DistributeData(dataChannels *dataSource.DataChannels) {
|
|
|
|
//数据类型注册
|
|
gob.Register([]interface{}{})
|
|
|
|
for {
|
|
if len(the.nodeList) == 0 {
|
|
log.Printf("nodeList is empty!")
|
|
time.Sleep(time.Second * 10)
|
|
continue
|
|
}
|
|
|
|
select {
|
|
case stopEnable := <-the.sleepCH:
|
|
if stopEnable {
|
|
stopTime := time.Second * 10
|
|
log.Printf("node 处理积压,%v,master 暂停 %v", stopEnable, stopTime)
|
|
time.Sleep(stopTime)
|
|
} else {
|
|
log.Printf("node 处理积压,%v,不正常空数据", stopEnable)
|
|
}
|
|
default:
|
|
}
|
|
|
|
select {
|
|
case args := <-dataChannels.RawDataChan:
|
|
the.notifyRawData("etNode", args)
|
|
case args := <-dataChannels.AggDataChan:
|
|
the.notifyAggData("etAgg", args)
|
|
}
|
|
}
|
|
}
|
|
func contains(arr []string, target string) bool {
|
|
for _, value := range arr {
|
|
if value == target {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
func (the *EtMaster) sortNodeListByThingCount() {
|
|
sort.Slice(the.nodeList, func(i, j int) bool {
|
|
return len(the.nodeList[i].args.ThingIds) < len(the.nodeList[j].args.ThingIds)
|
|
})
|
|
}
|
|
func (the *EtMaster) notifyRawData(nodeType string, d common_models.IotaData) {
|
|
|
|
isMatch := false
|
|
for _, nodeRpc := range the.nodeList {
|
|
if contains(nodeRpc.args.ThingIds, d.ThingId) {
|
|
isMatch = true
|
|
switch nodeRpc.args.NodeType {
|
|
case nodeType:
|
|
the.call_etNode(nodeRpc, &d)
|
|
}
|
|
}
|
|
}
|
|
//无匹配触发 reBalance
|
|
if !isMatch {
|
|
if len(the.nodeList) > 0 {
|
|
the.sortNodeListByThingCount()
|
|
the.nodeList[0].args.ThingIds = append(the.nodeList[0].args.ThingIds, d.ThingId)
|
|
log.Printf("thingId:[%s] 分配到node:[%s]", d.ThingId, the.nodeList[0].args.Addr)
|
|
the.call_etNode(the.nodeList[0], &d)
|
|
}
|
|
}
|
|
}
|
|
func (the *EtMaster) notifyAggData(nodeType string, d common_models.AggData) {
|
|
for _, nodeRpc := range the.nodeList {
|
|
switch nodeRpc.args.NodeType {
|
|
case nodeType:
|
|
go the.call_aggNode(nodeRpc, &d)
|
|
}
|
|
}
|
|
}
|
|
|
|
func timeCost(nodeId, deviceId string, start time.Time) {
|
|
tc := time.Since(start)
|
|
log.Printf("调用node[%s],[%s]耗时 = %v", nodeId, deviceId, tc)
|
|
}
|
|
|