zeaslity/ProjectOctopus
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

[ Cmii ] [ Octopus ] - reformat agent-go - 2

Browse Source
This commit is contained in:
zeaslity
2024-03-29 15:04:27 +08:00
parent 1d32d7dd04
commit 77c689abd7
6 changed files with 128 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
agent-go/a_agent/AgentHandler.go
View File

@@ -1,9 +1,25 @@
package a_agent
import "wdd.io/agent-common/logger"
import (
"wdd.io/agent-common/logger"
"wdd.io/agent-go/rabbitmq"
)
var log = logger.Log
func Activate() {
log.Info("Module [ AGENT ] activated !")
for {
if octopusMessage, ok := <-rabbitmq.BusinessRuntimeQueue.ReceiveChan.ExecutorRChan; ok {
// 处理数
// 输出日志
log.InfoF("接收到查询Agent状态的请求，结果为 => %s", octopusMessage)
} else {
// channel已关闭，跳出循环
log.ErrorF("business queue [ AGENT ] receive chan has closed !")
break
}
}
}

71
agent-go/a_executor/CommandExecutor.go
View File

@@ -3,11 +3,14 @@ package a_executor
import (
"bufio"
"bytes"
"encoding/json"
"fmt"
"os/exec"
"strconv"
"strings"
"wdd.io/agent-common/logger"
"wdd.io/agent-common/utils"
"wdd.io/agent-go/rabbitmq"
)
type ExecutionMessage struct {
@@ -28,35 +31,45 @@ var AgentOsOperatorCache = &AgentOsOperator{}
func Activate() {
log.Info("Module [ EXECUTOR ] activated !")
//// 转换类型
//executionMsgString := octopusMessage.Content.(string)
//
//// 解析 ExecutionMessage
//var executionMessage *a_executor.ExecutionMessage
//err := json.Unmarshal([]byte(executionMsgString), &executionMessage)
//if err != nil {
// log.Error(fmt.Sprintf("execution message convert to json is wrong! msg is => %s", executionMsgString))
// return
//}
//
//// 执行命令
//ok, resultLog := a_executor.Execute(executionMessage)
//if ok {
// octopusMessage.ResultCode = "200"
//} else {
// octopusMessage.ResultCode = "300"
//}
//
//// 返回结果
//if executionMessage.NeedResultReplay {
// // send back the result log
// octopusMessage.Result = resultLog
//}
//// 返回时间
//octopusMessage.ACTime = utils.ParseDateTimeTime()
//
//// 返回结果
//octopusMessage.SendToOctopusServer()
// 死循环获取channel中的数据
for {
if octopusMessage, ok := <-rabbitmq.BusinessRuntimeQueue.ReceiveChan.ExecutorRChan; ok {
// 处理数据
// 转换类型
executionMsgString := octopusMessage.Content.(string)
//解析 ExecutionMessage
var executionMessage *ExecutionMessage
err := json.Unmarshal([]byte(executionMsgString), &executionMessage)
if err != nil {
log.Error(fmt.Sprintf("execution message convert to json is wrong! msg is => %s", executionMsgString))
return
}
// 执行命令
ok, resultLog := Execute(executionMessage)
if ok {
octopusMessage.ResultCode = "200"
} else {
octopusMessage.ResultCode = "300"
}
// 返回结果
if executionMessage.NeedResultReplay {
// send back the result log
octopusMessage.Result = resultLog
}
// 返回时间
octopusMessage.ACTime = utils.ParseDateTimeTime()
// 返回结果
octopusMessage.SendToOctopusServer()
} else {
// channel已关闭，跳出循环
log.ErrorF("business queue [ EXECUTOR ] receive chan has closed !")
break
}
}
}
func Execute(em *ExecutionMessage) (bool, []string) {

20
agent-go/a_init/AgentInitialization.go
View File

@@ -22,8 +22,6 @@ var P = g.G.P
var log = logger.Log
//var AgentServerInfoCache = &register.AgentServerInfo{}
func INIT(octopusAgentConfigFileName string, agentServerInfoConf string) chan bool {
// 获取系统的环境变量
@@ -153,6 +151,9 @@ func buildAndStartBusinessRuntimeQueue(agentTopicName string) chan bool {
// 开始处理Runtime的OM消息
businessForeverChan := octopusMsgQueue.Handle()
// cache it
rabbitmq.BusinessRuntimeQueue = octopusMsgQueue
return businessForeverChan
}
@@ -160,11 +161,20 @@ func buildAndStartBusinessRuntimeQueue(agentTopicName string) chan bool {
func activatedOctopusAgentModules() {
// Agent
a_agent.Activate()
err := P.Submit(func() {
a_agent.Activate()
})
if err != nil {
return
}
// Executor
a_executor.Activate()
_ = P.Submit(func() {
a_executor.Activate()
})
// Status
a_status.Activate()
_ = P.Submit(func() {
a_status.Activate()
})
}
func buildAndSendInitMsgToServer(agentServerInfo *a_agent.AgentServerInfo, initToServerQueue *rabbitmq.RabbitQueue) {

91
agent-go/a_status/Status.go
View File

@@ -1,10 +1,14 @@
package a_status
import (
"encoding/json"
"fmt"
"strings"
"time"
"wdd.io/agent-common/logger"
"wdd.io/agent-common/utils"
"wdd.io/agent-go/g"
"wdd.io/agent-go/rabbitmq"
)
var log = logger.Log
@@ -41,46 +45,53 @@ type AgentInfo struct {
func Activate() {
log.Info("Module [ STATUS ] activated !")
//v, ok := (octopusMessage.Content).(string)
//if !ok {
// log.ErrorF("convert to string is wrong %s", v)
//}
//
//statusMsgString := octopusMessage.Content.(string)
//
//var statusMessage *a_status.StatusMessage
//err := json.Unmarshal([]byte(statusMsgString), &statusMessage)
//if err != nil {
// fmt.Println(err.Error())
// log.Error(fmt.Sprintf("status message convert to json is wrong! msg is => %s", octopusMessage))
// return
//}
//
//// OMessageStatusTypeEnum
//var statusRes string
//if strings.HasPrefix(statusMessage.StatusType, "PING") {
// // ping info
// statusRes = a_status.Ping()
//} else if strings.HasPrefix(statusMessage.StatusType, "METRIC") {
// // metric info
// agentStatusString, _ := json.Marshal(a_status.ReportAgentMetric())
// statusRes = string(agentStatusString)
//} else if strings.HasPrefix(statusMessage.StatusType, "INFO") {
// log.InfoF("[statusOMHandler] - call for agent info !")
//} else {
// log.WarnF("[statusOMHandler] - error octopus status message type of %s", statusMessage.StatusType)
//}
//
//// 返回消息
//// 组装消息
//octopusMessage.ACTime = utils.ParseDateTimeTime()
//octopusMessage.Result = statusRes
//// 发送回去
//statusOctopusReplayMessage, _ := json.Marshal(octopusMessage)
//OctopusToServerQueue.Send(statusOctopusReplayMessage)
//
//// 输出日志
//log.InfoF("接收到查询Agent状态的请求，结果为 => %s", statusRes)
// 死循环获取channel中的数据
for {
if octopusMessage, ok := <-rabbitmq.BusinessRuntimeQueue.ReceiveChan.ExecutorRChan; ok {
// 处理数据
statusMsgString := octopusMessage.Content.(string)
var statusMessage *StatusMessage
err := json.Unmarshal([]byte(statusMsgString), &statusMessage)
if err != nil {
fmt.Println(err.Error())
log.Error(fmt.Sprintf("status message convert to json is wrong! msg is => %s", octopusMessage))
return
}
// OMessageStatusTypeEnum
var statusRes string
if strings.HasPrefix(statusMessage.StatusType, "PING") {
// ping info
statusRes = Ping()
} else if strings.HasPrefix(statusMessage.StatusType, "METRIC") {
// metric info
agentStatusString, _ := json.Marshal(ReportAgentMetric())
statusRes = string(agentStatusString)
} else if strings.HasPrefix(statusMessage.StatusType, "INFO") {
log.InfoF("[statusOMHandler] - call for agent info !")
} else {
log.WarnF("[statusOMHandler] - error octopus status message type of %s", statusMessage.StatusType)
}
// 返回消息
// 组装消息
octopusMessage.ACTime = utils.ParseDateTimeTime()
octopusMessage.Result = statusRes
// 发送回去
octopusMessage.SendToOctopusServer()
// 输出日志
log.InfoF("接收到查询Agent状态的请求，结果为 => %s", statusRes)
} else {
// channel已关闭，跳出循环
log.ErrorF("business queue [ STATUS ] receive chan has closed !")
break
}
}
}
func Ping() string {

2
agent-go/rabbitmq/RabbitMsgQueue.go
View File

@@ -9,6 +9,8 @@ import (
"wdd.io/agent-common/logger"
)
var BusinessRuntimeQueue = &RabbitQueue{}
type RabbitMQ interface {
RabbitSendWriter
RabbitConnectCloser