floraachy
/
console
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix discovery

This commit is contained in:
medcl 2023-10-24 23:28:24 +08:00
parent 6672f545b7
commit 7fa78f47ec
2 changed files with 25 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
modules/agent/api/elasticsearch.go
View File

@ -108,10 +108,17 @@ func refreshNodesInfo(inst *model.Instance) (*elastic.DiscoveryResult, error) {
} }
} }
var findPIDS=map[int]*elastic.NodesInfo{}
//TODO, merge requests to one //TODO, merge requests to one
for k, v := range enrolledNodesByAgent { for k, v := range enrolledNodesByAgent {
if _, ok := newNodes[k]; !ok { if _, ok := newNodes[k]; !ok {
client := elastic.GetClient(v.ClusterID) client := elastic.GetClientNoPanic(v.ClusterID)
if client == nil {
log.Error("client not found:", v.ClusterID)
continue
}
status := "online" status := "online"
nodeInfo, err := client.GetNodeInfo(v.NodeUUID) nodeInfo, err := client.GetNodeInfo(v.NodeUUID)
var clusterInfo *elastic.ClusterInformation var clusterInfo *elastic.ClusterInformation
@ -133,8 +140,6 @@ func refreshNodesInfo(inst *model.Instance) (*elastic.DiscoveryResult, error) {
log.Error("cluster info not found:",v.ClusterID,",",err,clusterInfo==nil) log.Error("cluster info not found:",v.ClusterID,",",err,clusterInfo==nil)
continue continue
} }
}else{ }else{
clusterInfo, err = adapter.ClusterVersion(elastic.GetMetadata(v.ClusterID)) clusterInfo, err = adapter.ClusterVersion(elastic.GetMetadata(v.ClusterID))
if err != nil || clusterInfo == nil{ if err != nil || clusterInfo == nil{
@ -143,6 +148,8 @@ func refreshNodesInfo(inst *model.Instance) (*elastic.DiscoveryResult, error) {
} }
} }
findPIDS[nodeInfo.Process.Id]=nodeInfo
newNodes[k] = &elastic.LocalNodeInfo{ newNodes[k] = &elastic.LocalNodeInfo{
Status: status, Status: status,
ClusterID: v.ClusterID, ClusterID: v.ClusterID,
@ -155,7 +162,14 @@ func refreshNodesInfo(inst *model.Instance) (*elastic.DiscoveryResult, error) {
} }
nodesInfo.Nodes = newNodes nodesInfo.Nodes = newNodes
newUnknows:=[]model.ProcessInfo{}
for _,v:=range nodesInfo.UnknownProcess{
if _,ok:=findPIDS[v.PID];!ok{
newUnknows=append(newUnknows,v)
}
}
nodesInfo.UnknownProcess=newUnknows
return nodesInfo, nil return nodesInfo, nil
} }
@ -362,6 +376,10 @@ type ClusterInfo struct {
ClusterIDs []string `json:"cluster_id"` ClusterIDs []string `json:"cluster_id"`
} }
func (h *APIHandler) autoEnrollESNode(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
//TODO
}
func (h *APIHandler) discoveryESNodesInfo(w http.ResponseWriter, req *http.Request, ps httprouter.Params) { func (h *APIHandler) discoveryESNodesInfo(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
id := ps.MustGetParameter("instance_id") id := ps.MustGetParameter("instance_id")
@ -385,12 +403,12 @@ func (h *APIHandler) discoveryESNodesInfo(w http.ResponseWriter, req *http.Reque
if len(nodes.UnknownProcess) > 0 { if len(nodes.UnknownProcess) > 0 {
var discoveredPIDs map[int]*elastic.LocalNodeInfo = make(map[int]*elastic.LocalNodeInfo) var discoveredPIDs map[int]*elastic.LocalNodeInfo = make(map[int]*elastic.LocalNodeInfo)
if req.Method == "POST" { if req.Method == "POST" {
bytes, err := h.GetRawBody(req) bytes, err := h.GetRawBody(req)
if err != nil { if err != nil {
panic(err) panic(err)
} }
if len(bytes) > 0 { if len(bytes) > 0 {
clusterInfo := ClusterInfo{} clusterInfo := ClusterInfo{}
util.FromJSONBytes(bytes, &clusterInfo) util.FromJSONBytes(bytes, &clusterInfo)
@ -418,8 +436,9 @@ func (h *APIHandler) discoveryESNodesInfo(w http.ResponseWriter, req *http.Reque
//try https again //try https again
success, tryAgain, nodeInfo = h.getESNodeInfoViaProxy(nodeHost, "https", auth, &instance) success, tryAgain, nodeInfo = h.getESNodeInfoViaProxy(nodeHost, "https", auth, &instance)
} }
if success { if success {
log.Error("connect to es node success:", nodeHost) log.Debug("connect to es node success:", nodeHost,", pid: ",node.PID)
discoveredPIDs[node.PID] = nodeInfo discoveredPIDs[node.PID] = nodeInfo
break break
} }

1
modules/agent/api/init.go
View File

@ -24,6 +24,7 @@ func Init() {
//bind agent with nodes //bind agent with nodes
api.HandleAPIMethod(api.GET, "/instance/:instance_id/node/_discovery", handler.RequirePermission(handler.discoveryESNodesInfo, enum.PermissionAgentInstanceRead)) api.HandleAPIMethod(api.GET, "/instance/:instance_id/node/_discovery", handler.RequirePermission(handler.discoveryESNodesInfo, enum.PermissionAgentInstanceRead))
api.HandleAPIMethod(api.POST, "/instance/:instance_id/node/_discovery", handler.RequirePermission(handler.discoveryESNodesInfo, enum.PermissionAgentInstanceRead)) api.HandleAPIMethod(api.POST, "/instance/:instance_id/node/_discovery", handler.RequirePermission(handler.discoveryESNodesInfo, enum.PermissionAgentInstanceRead))
api.HandleAPIMethod(api.POST, "/instance/:instance_id/node/_auto_enroll", handler.RequirePermission(handler.autoEnrollESNode, enum.PermissionAgentInstanceWrite))
api.HandleAPIMethod(api.POST, "/instance/:instance_id/node/_enroll", handler.RequirePermission(handler.enrollESNode, enum.PermissionAgentInstanceWrite)) api.HandleAPIMethod(api.POST, "/instance/:instance_id/node/_enroll", handler.RequirePermission(handler.enrollESNode, enum.PermissionAgentInstanceWrite))
api.HandleAPIMethod(api.POST, "/instance/:instance_id/node/_revoke", handler.RequirePermission(handler.revokeESNode, enum.PermissionAgentInstanceWrite)) api.HandleAPIMethod(api.POST, "/instance/:instance_id/node/_revoke", handler.RequirePermission(handler.revokeESNode, enum.PermissionAgentInstanceWrite))