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

cleanup unused code

This commit is contained in:
medcl 2023-10-20 17:14:14 +08:00
parent 05e1efa197
commit bea9ca6d1c
8 changed files with 459 additions and 1651 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -7,16 +7,21 @@ package api
import (
"bytes"
"context"
"errors"
"fmt"
log "github.com/cihub/seelog"
httprouter "infini.sh/framework/core/api/router"
"infini.sh/framework/core/elastic"
"infini.sh/framework/core/global"
"infini.sh/framework/core/model"
"infini.sh/framework/core/orm"
"infini.sh/framework/core/util"
common2 "infini.sh/framework/modules/elastic/common"
"infini.sh/framework/plugins/managed/common"
"infini.sh/framework/plugins/managed/server"
"net/http"
"net/url"
"sync"
"time"
)
@ -91,7 +96,6 @@ func refreshNodesInfo(inst *model.Instance) (*elastic.DiscoveryResult, error) {
}
for nodeID, node := range nodesInfo.Nodes {
v, ok := enrolledNodesByAgent[nodeID]
if ok {
node.ClusterID = v.ClusterID
@ -99,188 +103,14 @@ func refreshNodesInfo(inst *model.Instance) (*elastic.DiscoveryResult, error) {
}
}
////not recognized by agent, need auth?
//for _, node := range nodesInfo.UnknownProcess{
// for _, v := range node.ListenAddresses {
// //ask user to manual enroll this node
// //check local credentials, if it works, get node info
// }
//}
// {
// //node was not recognized by agent, need auth?
// if node.HttpPort != "" {
// for _, v := range enrolledNodesByAgent {
// if v.PublishAddress != "" {
// if util.UnifyLocalAddress(v.PublishAddress) == util.UnifyLocalAddress(node.PublishAddress) {
// node.ClusterID = v.ClusterID
// node.ClusterName = v.ClusterName
// node.NodeUUID = v.NodeUUID
// node.ClusterUuid = v.ClusterUUID
// node.NodeName = v.NodeName
// node.Path.Home = v.PathHome
// node.Path.Logs = v.PathLogs
// node.AgentID = inst.ID
// //TODO verify node info if the node id really match, need to fetch the credentials for agent
// //or let manager sync configs to this agent, verify the node info after receiving the configs
// //report any error along with this agent and node info
// break
// }
// }
// }
// }
//
//}
return nodesInfo, nil
}
type RemoteConfig struct {
orm.ORMObjectBase
Metadata model.Metadata `json:"metadata" elastic_mapping:"metadata: { type: object }"`
Payload common.ConfigFile `json:"payload" elastic_mapping:"payload: { type: object}"`
}
func remoteConfigProvider(instance model.Instance) []*common.ConfigFile {
//fetch configs from remote db
//fetch configs assigned to (instance=_all OR instance=$instance_id ) AND application.name=$application.name
q := orm.Query{
Size: 1000,
Conds: orm.And(orm.Eq("metadata.category", "app_settings"),
orm.Eq("metadata.name", instance.Application.Name),
orm.Eq("metadata.labels.instance", "_all"),
),
}
err, searchResult := orm.Search(RemoteConfig{}, &q)
if err != nil {
panic(err)
}
result := []*common.ConfigFile{}
for _, row := range searchResult.Result {
v, ok := row.(map[string]interface{})
if ok {
x, ok := v["payload"]
if ok {
f, ok := x.(map[string]interface{})
if ok {
name, ok := f["name"].(string)
if ok {
item := common.ConfigFile{}
item.Name = util.ToString(name)
item.Location = util.ToString(f["location"])
item.Content = util.ToString(f["content"])
item.Version,_ = util.ToInt64(util.ToString(f["version"]))
item.Size = int64(len(item.Content))
item.Managed = true
t, ok := v["updated"]
if ok {
layout := "2006-01-02T15:04:05.999999-07:00"
t1, err := time.Parse(layout, util.ToString(t))
if err == nil {
item.Updated = t1.Unix()
}
}
result=append(result,&item)
}
}
}
}
}
return result
}
func dynamicAgentConfigProvider(instance model.Instance) []*common.ConfigFile {
//get config files from remote db
//get settings with this agent id
result := []*common.ConfigFile{}
ids, err := GetEnrolledNodesByAgent(&instance)
if err != nil {
panic(err)
}
var latestTimestamp int64
for _, v := range ids {
if v.Updated > latestTimestamp {
latestTimestamp = v.Updated
}
}
if len(ids) > 0 {
cfg := common.ConfigFile{}
cfg.Name = "generated_metrics_tasks.yml"
cfg.Location = "generated_metrics_tasks.yml"
cfg.Content = getConfigs(ids)
cfg.Size = int64(len(cfg.Content))
cfg.Version = latestTimestamp
cfg.Managed = true
cfg.Updated = latestTimestamp
result = append(result, &cfg)
}
return result
}
func getConfigs(items map[string]BindingItem) string {
buffer := bytes.NewBuffer([]byte("configs.template:\n "))
for _, v := range items {
if v.ClusterID == "" {
panic("cluster id is empty")
}
metadata := elastic.GetMetadata(v.ClusterID)
var clusterLevelEnabled = false
var nodeLevelEnabled = true
var clusterEndPoint = metadata.Config.GetAnyEndpoint()
credential, err := common2.GetCredential(metadata.Config.CredentialID)
if err != nil {
panic(err)
}
var dv interface{}
dv, err = credential.Decode()
if err != nil {
panic(err)
}
var username = ""
var password = ""
if auth, ok := dv.(model.BasicAuth); ok {
username = auth.Username
password = auth.Password
}
buffer.Write([]byte(fmt.Sprintf("- name: \"%v\"\n path: ./config/task_config.tpl\n "+
"variable:\n "+
"CLUSTER_ID: %v\n "+
"CLUSTER_ENDPOINT: [\"%v\"]\n "+
"CLUSTER_USERNAME: \"%v\"\n "+
"CLUSTER_PASSWORD: \"%v\"\n "+
"CLUSTER_LEVEL_TASKS_ENABLED: %v\n "+
"NODE_LEVEL_TASKS_ENABLED: %v\n "+
"NODE_LOGS_PATH: \"%v\"\n\n\n"+
"#MANAGED_CONFIG_VERSION: %v\n"+
"#MANAGED: true",
v.NodeUUID, v.ClusterID, clusterEndPoint, username, password, clusterLevelEnabled, nodeLevelEnabled, v.PathLogs, v.Updated)))
}
//password: $[[keystore.$[[CLUSTER_ID]]_password]]
return buffer.String()
}
//get nodes info via agent
func GetElasticsearchNodesViaAgent(ctx context.Context, instance *model.Instance) (*elastic.DiscoveryResult, error) {
req := &util.Request{
Method: http.MethodGet,
Path: "/elasticsearch/nodes/_discovery",
Path: "/elasticsearch/node/_discovery",
Context: ctx,
}
@ -293,37 +123,6 @@ func GetElasticsearchNodesViaAgent(ctx context.Context, instance *model.Instance
return &obj, nil
}
func AuthESNode(ctx context.Context, agentBaseURL string, cfg elastic.ElasticsearchConfig) (*model.ESNodeInfo, error) {
reqBody, err := util.ToJSONBytes(cfg)
if err != nil {
return nil, err
}
req := &util.Request{
Method: http.MethodPost,
Path: "/elasticsearch/_auth",
Context: ctx,
Body: reqBody,
}
resBody := &model.ESNodeInfo{}
err = DoRequest(req, resBody)
if err != nil {
return nil, err
}
return resBody, nil
}
func getNodeByPidOrUUID(nodes map[int]*model.ESNodeInfo, pid int, uuid string, port string) *model.ESNodeInfo {
if nodes[pid] != nil {
return nodes[pid]
}
for _, node := range nodes {
if node.NodeUUID != "" && node.NodeUUID == uuid {
return node
}
}
return nil
}
type BindingItem struct {
ClusterName string `json:"cluster_name"`
ClusterUUID string `json:"cluster_uuid"`
@ -338,50 +137,19 @@ type BindingItem struct {
Updated int64 `json:"updated"`
}
func getUnAssociateNodes() (map[string][]model.ESNodeInfo, error) {
query := util.MapStr{
"size": 3000,
"query": util.MapStr{
"bool": util.MapStr{
"must_not": []util.MapStr{
{
"exists": util.MapStr{
"field": "cluster_id",
},
},
},
},
},
}
q := orm.Query{
RawQuery: util.MustToJSONBytes(query),
}
err, result := orm.Search(model.ESNodeInfo{}, &q)
if err != nil {
return nil, err
}
nodesInfo := map[string][]model.ESNodeInfo{}
for _, row := range result.Result {
node := model.ESNodeInfo{}
buf := util.MustToJSONBytes(row)
util.MustFromJSONBytes(buf, &node)
nodesInfo[node.AgentID] = append(nodesInfo[node.AgentID], node)
}
return nodesInfo, nil
}
func GetElasticLogFiles(ctx context.Context, instance *model.Instance, logsPath string) (interface{}, error) {
reqBody := util.MustToJSONBytes(util.MapStr{
"logs_path": logsPath,
})
req := &util.Request{
Method: http.MethodPost,
Path: "/elasticsearch/logs/_list",
Context: ctx,
Body: reqBody,
}
resBody := map[string]interface{}{}
err := doRequest(instance, req, &resBody)
if err != nil {
@ -391,6 +159,7 @@ func GetElasticLogFiles(ctx context.Context, instance *model.Instance, logsPath
return nil, fmt.Errorf("get elasticsearch log files error: %v", resBody)
}
return resBody["result"], nil
}
func GetElasticLogFileContent(ctx context.Context, instance *model.Instance, body interface{}) (interface{}, error) {
@ -535,3 +304,281 @@ func getAgentByNodeID(nodeID string) (*model.Instance, string, error) {
}
return nil, "", nil
}
type ClusterInfo struct {
ClusterIDs []string `json:"cluster_id"`
}
func (h *APIHandler) discoveryESNodesInfo(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
id := ps.MustGetParameter("instance_id")
instance := model.Instance{}
instance.ID = id
exists, err := orm.Get(&instance)
if !exists || err != nil {
h.WriteJSON(w, util.MapStr{
"_id": id,
"found": false,
}, http.StatusNotFound)
return
}
nodes, err := refreshNodesInfo(&instance)
if err != nil {
h.WriteError(w, err.Error(), http.StatusInternalServerError)
return
}
if len(nodes.UnknownProcess) > 0 {
var discoveredPIDs map[int]*elastic.LocalNodeInfo = make(map[int]*elastic.LocalNodeInfo)
if req.Method == "POST" {
bytes, err := h.GetRawBody(req)
if err != nil {
panic(err)
}
if len(bytes) > 0 {
clusterInfo := ClusterInfo{}
util.FromJSONBytes(bytes, &clusterInfo)
if len(clusterInfo.ClusterIDs) > 0 {
//try connect this node to cluster by using this cluster's agent credential
for _, clusterID := range clusterInfo.ClusterIDs {
meta := elastic.GetMetadata(clusterID)
if meta != nil {
if meta.Config.AgentCredentialID != "" {
auth, err := common2.GetAgentBasicAuth(meta.Config)
if err != nil {
panic(err)
}
if auth != nil {
//try connect
for _, node := range nodes.UnknownProcess {
for _, v := range node.ListenAddresses {
ip := v.IP
if util.ContainStr(v.IP, "::") {
ip = fmt.Sprintf("[%s]", v.IP)
}
nodeHost := fmt.Sprintf("%s:%d", ip, v.Port)
success, tryAgain, nodeInfo := h.getESNodeInfoViaProxy(nodeHost, "http", auth, instance)
if !success && tryAgain {
//try https again
success, tryAgain, nodeInfo = h.getESNodeInfoViaProxy(nodeHost, "https", auth, instance)
}
if success {
log.Error("connect to es node success:", nodeHost)
discoveredPIDs[node.PID] = nodeInfo
break
}
}
}
}
}
}
}
}
}
}
newUnknownProcess := []model.ProcessInfo{}
if len(discoveredPIDs) > 0 {
for _, node := range nodes.UnknownProcess {
if item, ok := discoveredPIDs[node.PID]; !ok {
newUnknownProcess = append(newUnknownProcess, node)
} else {
nodes.Nodes[item.NodeUUID] = item
}
}
nodes.UnknownProcess = newUnknownProcess
}
}
h.WriteJSON(w, nodes, http.StatusOK)
}
func (h *APIHandler) getESNodeInfoViaProxy(host string, schema string, auth *model.BasicAuth, instance model.Instance) (success, tryAgain bool, info *elastic.LocalNodeInfo) {
esConfig := elastic.ElasticsearchConfig{Host: host, Schema: schema, BasicAuth: auth}
body := util.MustToJSONBytes(esConfig)
res, err := server.ProxyRequestToRuntimeInstance(instance.GetEndpoint(), "POST", "/elasticsearch/node/_info",
body, int64(len(body)), auth)
if err != nil {
panic(err)
}
if global.Env().IsDebug {
if res != nil { // && res.StatusCode==http.StatusOK
log.Debug(string(res.Body))
}
}
if res != nil && res.StatusCode == http.StatusForbidden {
return false, false, nil
}
if res != nil && res.StatusCode == http.StatusOK {
node := elastic.LocalNodeInfo{}
err := util.FromJSONBytes(res.Body, &node)
if err != nil {
panic(err)
}
return true, false, &node
}
return false, true, nil
}
func NewClusterSettings(clusterID string) *model.Setting {
settings := model.Setting{
Metadata: model.Metadata{
Category: Cluster,
},
}
settings.ID = fmt.Sprintf("%v_%v_%v", settings.Metadata.Category, settings.Metadata.Name, clusterID)
settings.Metadata.Labels = util.MapStr{
"cluster_id": clusterID,
}
return &settings
}
func NewNodeAgentSettings(instanceID string, item *BindingItem) *model.Setting {
settings := model.Setting{
Metadata: model.Metadata{
Category: Node,
Name: "agent",
},
}
settings.ID = fmt.Sprintf("%v_%v_%v", settings.Metadata.Category, settings.Metadata.Name, item.NodeUUID)
settings.Metadata.Labels = util.MapStr{
"agent_id": instanceID,
}
settings.Payload = util.MapStr{
"cluster_id": item.ClusterID,
"cluster_name": item.ClusterName,
"cluster_uuid": item.ClusterUUID,
"node_uuid": item.NodeUUID,
"publish_address": item.PublishAddress,
"node_name": item.NodeName,
"path_home": item.PathHome,
"path_logs": item.PathLogs,
}
return &settings
}
func NewIndexSettings(clusterID, nodeID, agentID, indexName, indexID string) *model.Setting {
settings := model.Setting{
Metadata: model.Metadata{
Category: Index,
},
}
settings.ID = fmt.Sprintf("%v_%v_%v", settings.Metadata.Category, settings.Metadata.Name, nodeID)
settings.Metadata.Labels = util.MapStr{
"cluster_id": clusterID,
"node_id": nodeID,
"agent_id": agentID,
"index_name": indexName,
"index_id": indexID,
}
return &settings
}
const Cluster = "cluster_settings"
const Node = "node_settings"
const Index = "index_settings"
func (h *APIHandler) revokeESNode(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
//agent id
instID := ps.MustGetParameter("instance_id")
item := BindingItem{}
err := h.DecodeJSON(req, &item)
if err != nil {
h.WriteError(w, err.Error(), http.StatusInternalServerError)
return
}
settings := NewNodeAgentSettings(instID, &item)
err = orm.Delete(&orm.Context{
Refresh: "wait_for",
}, settings)
if err != nil {
h.WriteError(w, err.Error(), http.StatusInternalServerError)
return
}
h.WriteAckOKJSON(w)
}
func (h *APIHandler) enrollESNode(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
//agent id
instID := ps.MustGetParameter("instance_id")
//node id and cluster id
item := BindingItem{}
err := h.DecodeJSON(req, &item)
if err != nil {
h.WriteError(w, err.Error(), http.StatusInternalServerError)
return
}
//update node's setting
settings := NewNodeAgentSettings(instID, &item)
err = orm.Update(&orm.Context{
Refresh: "wait_for",
}, settings)
if err != nil {
h.WriteError(w, err.Error(), http.StatusInternalServerError)
return
}
h.WriteAckOKJSON(w)
}
var mTLSClient *http.Client //TODO get mTLSClient
var initOnce = sync.Once{}
func doRequest(instance *model.Instance, req *util.Request, obj interface{}) error {
var err error
var res *util.Result
initOnce.Do(func() {
if global.Env().SystemConfig.Configs.TLSConfig.TLSEnabled && global.Env().SystemConfig.Configs.TLSConfig.TLSCAFile != "" {
//init client
hClient, err := util.NewMTLSClient(
global.Env().SystemConfig.Configs.TLSConfig.TLSCAFile,
global.Env().SystemConfig.Configs.TLSConfig.TLSCertFile,
global.Env().SystemConfig.Configs.TLSConfig.TLSKeyFile)
if err != nil {
panic(err)
}
mTLSClient = hClient
}
})
req.Url, err = url.JoinPath(instance.GetEndpoint(), req.Path)
res, err = util.ExecuteRequestWithCatchFlag(mTLSClient, req, true)
if err != nil || res.StatusCode != 200 {
body := ""
if res != nil {
body = string(res.Body)
}
return errors.New(fmt.Sprintf("request error: %v, %v", err, body))
}
if res != nil {
if res.Body != nil {
return util.FromJSONBytes(res.Body, obj)
}
}
return nil
}

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

@ -10,6 +10,10 @@ import (
"infini.sh/framework/plugins/managed/server"
)
type APIHandler struct {
api.Handler
}
func Init() {
handler := APIHandler{}
api.HandleAPIMethod(api.POST, "/host/_enroll", handler.enrollHost)
@ -18,23 +22,17 @@ func Init() {
api.HandleAPIMethod(api.DELETE, "/host/:host_id", handler.deleteHost)
//bind agent with nodes
api.HandleAPIMethod(api.GET, "/instance/:instance_id/node/_discovery", handler.RequirePermission(handler.getESNodesInfo, 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/_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, "/auto_associate", handler.RequirePermission(handler.autoAssociateESNode, enum.PermissionAgentInstanceWrite))
//api.HandleAPIMethod(api.POST, "/instance/:instance_id/node/_auth", handler.RequirePermission(handler.authESNode, enum.PermissionAgentInstanceWrite))
//api.HandleAPIMethod(api.DELETE, "/instance/:instance_id/_nodes", handler.RequirePermission(handler.deleteESNode, enum.PermissionAgentInstanceWrite))
//api.HandleAPIMethod(api.POST, "/auto_associate", handler.RequirePermission(handler.autoAssociateESNode, enum.PermissionAgentInstanceWrite))
//get elasticsearch node logs, direct fetch or via stored logs(TODO)
api.HandleAPIMethod(api.GET, "/elasticsearch/:id/node/:node_id/logs/_list", handler.RequirePermission(handler.getLogFilesByNode, enum.PermissionAgentInstanceRead))
api.HandleAPIMethod(api.POST, "/elasticsearch/:id/node/:node_id/logs/_read", handler.RequirePermission(handler.getLogFileContent, enum.PermissionAgentInstanceRead))
//api.HandleAPIMethod(api.POST, "/agent/install_command", handler.RequireLogin(handler.generateInstallCommand))
//api.HandleAPIMethod(api.GET, "/agent/install.sh", handler.getInstallScript)
server.RegisterConfigProvider(remoteConfigProvider)
server.RegisterConfigProvider(dynamicAgentConfigProvider)
}

14
modules/agent/api/instance.go
View File

@ -1,14 +0,0 @@
/* Copyright © INFINI Ltd. All rights reserved.
* Web: https://infinilabs.com
* Email: hello#infini.ltd */
package api
import (
"infini.sh/framework/core/api"
)
type APIHandler struct {
api.Handler
}

165
modules/agent/api/remote_config.go Normal file
View File

@ -0,0 +1,165 @@
/* Copyright © INFINI LTD. All rights reserved.
* Web: https://infinilabs.com
* Email: hello#infini.ltd */
package api
import (
"bytes"
"fmt"
"infini.sh/framework/core/elastic"
"infini.sh/framework/core/model"
"infini.sh/framework/core/orm"
"infini.sh/framework/core/util"
common2 "infini.sh/framework/modules/elastic/common"
"infini.sh/framework/plugins/managed/common"
"time"
)
type RemoteConfig struct {
orm.ORMObjectBase
Metadata model.Metadata `json:"metadata" elastic_mapping:"metadata: { type: object }"`
Payload common.ConfigFile `json:"payload" elastic_mapping:"payload: { type: object}"`
}
func remoteConfigProvider(instance model.Instance) []*common.ConfigFile {
//fetch configs from remote db
//fetch configs assigned to (instance=_all OR instance=$instance_id ) AND application.name=$application.name
q := orm.Query{
Size: 1000,
Conds: orm.And(orm.Eq("metadata.category", "app_settings"),
orm.Eq("metadata.name", instance.Application.Name),
orm.Eq("metadata.labels.instance", "_all"),
),
}
err, searchResult := orm.Search(RemoteConfig{}, &q)
if err != nil {
panic(err)
}
result := []*common.ConfigFile{}
for _, row := range searchResult.Result {
v, ok := row.(map[string]interface{})
if ok {
x, ok := v["payload"]
if ok {
f, ok := x.(map[string]interface{})
if ok {
name, ok := f["name"].(string)
if ok {
item := common.ConfigFile{}
item.Name = util.ToString(name)
item.Location = util.ToString(f["location"])
item.Content = util.ToString(f["content"])
item.Version, _ = util.ToInt64(util.ToString(f["version"]))
item.Size = int64(len(item.Content))
item.Managed = true
t, ok := v["updated"]
if ok {
layout := "2006-01-02T15:04:05.999999-07:00"
t1, err := time.Parse(layout, util.ToString(t))
if err == nil {
item.Updated = t1.Unix()
}
}
result = append(result, &item)
}
}
}
}
}
return result
}
func dynamicAgentConfigProvider(instance model.Instance) []*common.ConfigFile {
//get config files from remote db
//get settings with this agent id
result := []*common.ConfigFile{}
ids, err := GetEnrolledNodesByAgent(&instance)
if err != nil {
panic(err)
}
var latestTimestamp int64
for _, v := range ids {
if v.Updated > latestTimestamp {
latestTimestamp = v.Updated
}
}
if len(ids) > 0 {
cfg := common.ConfigFile{}
cfg.Name = "generated_metrics_tasks.yml"
cfg.Location = "generated_metrics_tasks.yml"
cfg.Content = getAgentIngestConfigs(ids)
cfg.Size = int64(len(cfg.Content))
cfg.Version = latestTimestamp
cfg.Managed = true
cfg.Updated = latestTimestamp
result = append(result, &cfg)
}
return result
}
func getAgentIngestConfigs(items map[string]BindingItem) string {
buffer := bytes.NewBuffer([]byte("configs.template:\n "))
for _, v := range items {
if v.ClusterID == "" {
panic("cluster id is empty")
}
metadata := elastic.GetMetadata(v.ClusterID)
var clusterLevelEnabled = false
var nodeLevelEnabled = true
var clusterEndPoint = metadata.Config.GetAnyEndpoint()
var username = ""
var password = ""
if metadata.Config.AgentCredentialID != "" {
credential, err := common2.GetCredential(metadata.Config.AgentCredentialID)
if err != nil {
panic(err)
}
var dv interface{}
dv, err = credential.Decode()
if err != nil {
panic(err)
}
if auth, ok := dv.(model.BasicAuth); ok {
username = auth.Username
password = auth.Password
}
}
buffer.Write([]byte(fmt.Sprintf("- name: \"%v\"\n path: ./config/task_config.tpl\n "+
"variable:\n "+
"CLUSTER_ID: %v\n "+
"CLUSTER_ENDPOINT: [\"%v\"]\n "+
"CLUSTER_USERNAME: \"%v\"\n "+
"CLUSTER_PASSWORD: \"%v\"\n "+
"CLUSTER_LEVEL_TASKS_ENABLED: %v\n "+
"NODE_LEVEL_TASKS_ENABLED: %v\n "+
"NODE_LOGS_PATH: \"%v\"\n\n\n"+
"#MANAGED_CONFIG_VERSION: %v\n"+
"#MANAGED: true",
v.NodeUUID, v.ClusterID, clusterEndPoint, username, password, clusterLevelEnabled, nodeLevelEnabled, v.PathLogs, v.Updated)))
}
//password: $[[keystore.$[[CLUSTER_ID]]_password]]
return buffer.String()
}

882
modules/agent/api/tod.go
View File

@ -1,882 +0,0 @@
/* Copyright © INFINI LTD. All rights reserved.
* Web: https://infinilabs.com
* Email: hello#infini.ltd */
package api
import (
"context"
"errors"
"fmt"
log "github.com/cihub/seelog"
common2 "infini.sh/console/modules/agent/common"
model3 "infini.sh/console/modules/agent/model"
httprouter "infini.sh/framework/core/api/router"
"infini.sh/framework/core/elastic"
"infini.sh/framework/core/global"
"infini.sh/framework/core/host"
"infini.sh/framework/core/model"
"infini.sh/framework/core/orm"
"infini.sh/framework/core/util"
"infini.sh/framework/modules/elastic/common"
"net/http"
"net/url"
"strings"
"sync"
)
//func (h *APIHandler) updateInstance(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
// id := ps.MustGetParameter("instance_id")
// oldInst := model.Instance{}
// oldInst.ID = id
// _, err := orm.Get(&oldInst)
// if err != nil {
// if err == elastic2.ErrNotFound {
// h.WriteJSON(w, util.MapStr{
// "_id": id,
// "result": "not_found",
// }, http.StatusNotFound)
// return
// }
// h.WriteError(w, err.Error(), http.StatusInternalServerError)
// log.Error(err)
// return
// }
//
// obj := model.Instance{}
// err = h.DecodeJSON(req, &obj)
// if err != nil {
// h.WriteError(w, err.Error(), http.StatusInternalServerError)
// log.Error(err)
// return
// }
//
// oldInst.Name = obj.Name
// oldInst.Endpoint = obj.Endpoint
// oldInst.Description = obj.Description
// oldInst.Tags = obj.Tags
// oldInst.BasicAuth = obj.BasicAuth
// err = orm.Update(&orm.Context{
// Refresh: "wait_for",
// }, &oldInst)
// if err != nil {
// h.WriteError(w, err.Error(), http.StatusInternalServerError)
// log.Error(err)
// return
// }
//
// h.WriteJSON(w, util.MapStr{
// "_id": obj.ID,
// "result": "updated",
// }, 200)
//}
//
//func (h *APIHandler) searchInstance(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
//
// var (
// keyword = h.GetParameterOrDefault(req, "keyword", "")
// //queryDSL = `{"query":{"bool":{"must":[%s]}}, "size": %d, "from": %d}`
// strSize = h.GetParameterOrDefault(req, "size", "20")
// strFrom = h.GetParameterOrDefault(req, "from", "0")
// )
//
// var (
// mustQ []interface{}
// )
//
// if keyword != "" {
// mustQ = append(mustQ, util.MapStr{
// "query_string": util.MapStr{
// "default_field": "*",
// "query": keyword,
// },
// })
// }
// size, _ := strconv.Atoi(strSize)
// if size <= 0 {
// size = 20
// }
// from, _ := strconv.Atoi(strFrom)
// if from < 0 {
// from = 0
// }
//
// queryDSL := util.MapStr{
// "size": size,
// "from": from,
// }
// if len(mustQ) > 0 {
// queryDSL["query"] = util.MapStr{
// "bool": util.MapStr{
// "must": mustQ,
// },
// }
// }
//
// q := orm.Query{}
// q.RawQuery = util.MustToJSONBytes(queryDSL)
//
// err, res := orm.Search(&model.Instance{}, &q)
// if err != nil {
// log.Error(err)
// h.WriteError(w, err.Error(), http.StatusInternalServerError)
// return
// }
//
// h.Write(w, res.Raw)
//}
func (h *APIHandler) getESNodesInfo(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
id := ps.MustGetParameter("instance_id")
obj := model.Instance{}
obj.ID = id
exists, err := orm.Get(&obj)
if !exists || err != nil {
h.WriteJSON(w, util.MapStr{
"_id": id,
"found": false,
}, http.StatusNotFound)
return
}
nodes, err := refreshNodesInfo(&obj)
if err != nil {
h.WriteError(w, err.Error(), http.StatusInternalServerError)
return
}
h.WriteJSON(w, nodes, http.StatusOK)
}
func (h *APIHandler) authESNode(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
id := ps.MustGetParameter("instance_id")
inst := model.Instance{}
inst.ID = id
exists, err := orm.Get(&inst)
if !exists || err != nil {
h.WriteJSON(w, util.MapStr{
"_id": id,
"found": false,
}, http.StatusNotFound)
return
}
reqBody := struct {
NodeID string `json:"node_id"`
ESConfig *elastic.ElasticsearchConfig `json:"es_config"`
}{}
err = h.DecodeJSON(req, &reqBody)
if err != nil {
log.Error(err)
h.WriteError(w, err.Error(), http.StatusInternalServerError)
return
}
//node id maybe is missing
if reqBody.NodeID != "" {
//verify the node id, if the node is is actually the node of the instance
oldNodeInfo := &model.ESNodeInfo{
ID: reqBody.NodeID,
}
exists, err = orm.Get(oldNodeInfo)
if !exists || err != nil {
h.WriteJSON(w, fmt.Sprintf("node [%s] of agent [%s] was not found", oldNodeInfo.ID, inst.Name), http.StatusInternalServerError)
return
}
} else {
//find out the node id with credentials
cfg := reqBody.ESConfig
if cfg.Endpoint == "" {
cfg.Endpoint = cfg.GetAnyEndpoint()
}
basicAuth, err := common.GetBasicAuth(cfg)
if err != nil {
log.Error(err)
h.WriteError(w, err.Error(), http.StatusInternalServerError)
return
}
cfg.BasicAuth = basicAuth
nodeInfo, err := AuthESNode(context.Background(), inst.GetEndpoint(), *cfg)
if err != nil {
log.Error(err)
h.WriteError(w, err.Error(), http.StatusInternalServerError)
return
}
//host, port, err := net.SplitHostPort(nodeInfo.PublishAddress)
//if err != nil {
// log.Error(err)
// h.WriteError(w, err.Error(), http.StatusInternalServerError)
// return
//}
//if !util.StringInArray(inst.Network.IP, host) && !net.ParseIP(host).IsLoopback() {
// h.WriteError(w, fmt.Sprintf("got node host %s not match any ip of %v", host, inst.Network.IP), http.StatusInternalServerError)
// return
//}
//if oldNodeInfo.HttpPort != port {
// h.WriteError(w, fmt.Sprintf("port mismatch, got: %s，expected: %s", port, oldNodeInfo.HttpPort), http.StatusInternalServerError)
// return
//}
//if oldNodeInfo.ProcessInfo.PID != nodeInfo.ProcessInfo.PID {
// h.WriteError(w, fmt.Sprintf("process id mismatch, got: %d，expected: %d", nodeInfo.ProcessInfo.PID, oldNodeInfo.ProcessInfo.PID), http.StatusInternalServerError)
// return
//}
reqBody.NodeID = nodeInfo.NodeUUID
}
//nodeInfo:=elastic.NodeConfig{}
//nodeInfo.ID = reqBody.NodeID
//nodeInfo.AgentID = inst.ID
//err = orm.Update(nil, nodeInfo) //update node's info and agent_id
//if err != nil {
// log.Error(err)
// h.WriteError(w, err.Error(), http.StatusInternalServerError)
// return
//}
//h.WriteJSON(w, nodeInfo, http.StatusOK)
}
func NewClusterSettings(clusterID string) *model.Setting {
settings := model.Setting{
Metadata: model.Metadata{
Category: Cluster,
},
}
settings.ID = fmt.Sprintf("%v_%v_%v", settings.Metadata.Category, settings.Metadata.Name, clusterID)
settings.Metadata.Labels = util.MapStr{
"cluster_id": clusterID,
}
return &settings
}
func NewNodeAgentSettings(instanceID string, item *BindingItem) *model.Setting {
settings := model.Setting{
Metadata: model.Metadata{
Category: Node,
Name: "agent",
},
}
settings.ID = fmt.Sprintf("%v_%v_%v", settings.Metadata.Category, settings.Metadata.Name, item.NodeUUID)
settings.Metadata.Labels = util.MapStr{
"agent_id": instanceID,
}
settings.Payload = util.MapStr{
"cluster_id": item.ClusterID,
"cluster_name": item.ClusterName,
"cluster_uuid": item.ClusterUUID,
"node_uuid": item.NodeUUID,
"publish_address": item.PublishAddress,
"node_name": item.NodeName,
"path_home": item.PathHome,
"path_logs": item.PathLogs,
}
return &settings
}
func NewIndexSettings(clusterID, nodeID, agentID, indexName, indexID string) *model.Setting {
settings := model.Setting{
Metadata: model.Metadata{
Category: Index,
},
}
settings.ID = fmt.Sprintf("%v_%v_%v", settings.Metadata.Category, settings.Metadata.Name, nodeID)
settings.Metadata.Labels = util.MapStr{
"cluster_id": clusterID,
"node_id": nodeID,
"agent_id": agentID,
"index_name": indexName,
"index_id": indexID,
}
return &settings
}
const Cluster = "cluster_settings"
const Node = "node_settings"
const Index = "index_settings"
func (h *APIHandler) revokeESNode(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
//agent id
instID := ps.MustGetParameter("instance_id")
item := BindingItem{}
err := h.DecodeJSON(req, &item)
if err != nil {
h.WriteError(w, err.Error(), http.StatusInternalServerError)
return
}
settings := NewNodeAgentSettings(instID, &item)
err = orm.Delete(&orm.Context{
Refresh: "wait_for",
}, settings)
if err != nil {
h.WriteError(w, err.Error(), http.StatusInternalServerError)
return
}
h.WriteAckOKJSON(w)
}
func (h *APIHandler) enrollESNode(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
//agent id
instID := ps.MustGetParameter("instance_id")
//node id and cluster id
item := BindingItem{}
err := h.DecodeJSON(req, &item)
if err != nil {
h.WriteError(w, err.Error(), http.StatusInternalServerError)
return
}
//update node's setting
settings := NewNodeAgentSettings(instID, &item)
err = orm.Update(&orm.Context{
Refresh: "wait_for",
}, settings)
if err != nil {
h.WriteError(w, err.Error(), http.StatusInternalServerError)
return
}
h.WriteAckOKJSON(w)
}
func (h *APIHandler) autoAssociateESNode(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
reqBody := struct {
ClusterIDs []string `json:"cluster_ids"`
}{}
err := h.DecodeJSON(req, &reqBody)
if err != nil {
log.Error(err)
h.WriteError(w, err.Error(), http.StatusInternalServerError)
return
}
// query not associated nodes info
nodesM, err := getUnAssociateNodes()
if err != nil {
log.Error(err)
h.WriteError(w, err.Error(), http.StatusInternalServerError)
return
}
if len(nodesM) == 0 {
h.WriteAckOKJSON(w)
return
}
agentIds := make([]string, 0, len(nodesM))
for agentID := range nodesM {
agentIds = append(agentIds, agentID)
}
agents, err := getAgentByIds(agentIds)
if err != nil {
log.Error(err)
h.WriteError(w, err.Error(), http.StatusInternalServerError)
return
}
for _, clusterID := range reqBody.ClusterIDs {
// query cluster basicauth
cfg := elastic.GetConfig(clusterID)
basicAuth, err := common.GetBasicAuth(cfg)
if err != nil {
log.Error(err)
h.WriteError(w, err.Error(), http.StatusInternalServerError)
return
}
taskSetting, err := getSettingsByClusterID(cfg.ID)
if err != nil {
log.Error(err)
h.WriteError(w, err.Error(), http.StatusInternalServerError)
return
}
for agentID, nodes := range nodesM {
var (
inst *model.Instance
ok bool
)
if inst, ok = agents[agentID]; !ok {
log.Warnf("agent [%v] was not found", agentID)
continue
}
settings, err := common2.GetAgentSettings(agentID, 0)
if err != nil {
log.Error(err)
continue
}
for _, v := range nodes {
host := v.PublishAddress
var endpoint string
if strings.HasPrefix(host, "::") { //for ipv6
instURL, err := url.Parse(inst.Endpoint)
if err != nil {
log.Error(err)
continue
}
host = instURL.Hostname()
endpoint = fmt.Sprintf("%s://[%s]:%s", v.Schema, host, v.HttpPort)
} else {
endpoint = fmt.Sprintf("%s://%s", v.Schema, host)
}
escfg := elastic.ElasticsearchConfig{
Endpoint: endpoint,
BasicAuth: basicAuth,
}
nodeInfo, err := AuthESNode(context.Background(), inst.GetEndpoint(), escfg)
if err != nil {
log.Warn(err)
continue
}
//matched
if nodeInfo.ClusterUuid == cfg.ClusterUUID {
//update node info
nodeInfo.ID = v.ID
nodeInfo.AgentID = inst.ID
nodeInfo.ClusterID = cfg.ID
err = orm.Save(nil, nodeInfo)
if err != nil {
log.Error(err)
continue
}
setting := pickAgentSettings(settings, v)
if setting == nil {
tsetting := model3.TaskSetting{
NodeStats: &model3.NodeStatsTask{
Enabled: true,
},
Logs: &model3.LogsTask{
Enabled: true,
LogsPath: nodeInfo.Path.Logs,
},
}
if taskSetting.IndexStats != nil {
tsetting.IndexStats = taskSetting.IndexStats
taskSetting.IndexStats = nil
}
if taskSetting.ClusterHealth != nil {
tsetting.ClusterHealth = taskSetting.ClusterHealth
taskSetting.ClusterHealth = nil
}
if taskSetting.ClusterStats != nil {
tsetting.ClusterStats = taskSetting.ClusterStats
taskSetting.ClusterStats = nil
}
setting = &model.Setting{
Metadata: model.Metadata{
Category: "agent",
Name: "task",
Labels: util.MapStr{
"agent_id": agentID,
"cluster_uuid": nodeInfo.ClusterUuid,
"cluster_id": nodeInfo.ClusterID,
"node_uuid": nodeInfo.NodeUUID,
"endpoint": fmt.Sprintf("%s://%s", nodeInfo.Schema, nodeInfo.PublishAddress),
},
},
Payload: util.MapStr{
"task": tsetting,
},
}
err = orm.Create(nil, setting)
if err != nil {
log.Error("save agent task setting error: ", err)
h.WriteError(w, err.Error(), http.StatusInternalServerError)
return
}
}
}
}
}
}
h.WriteAckOKJSON(w)
}
func getAgentByIds(agentIDs []string) (map[string]*model.Instance, error) {
query := util.MapStr{
"size": len(agentIDs),
"query": util.MapStr{
"terms": util.MapStr{
"id": agentIDs,
},
},
}
q := orm.Query{
RawQuery: util.MustToJSONBytes(query),
}
err, result := orm.Search(model.Instance{}, &q)
if err != nil {
return nil, err
}
agents := map[string]*model.Instance{}
for _, row := range result.Result {
inst := model.Instance{}
buf := util.MustToJSONBytes(row)
util.MustFromJSONBytes(buf, &inst)
agents[inst.ID] = &inst
}
return agents, nil
}
func (h *APIHandler) deleteESNode(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
id := ps.MustGetParameter("instance_id")
nodeIDs := []string{}
err := h.DecodeJSON(req, &nodeIDs)
if err != nil {
log.Error(err)
h.WriteError(w, err.Error(), http.StatusInternalServerError)
return
}
if len(nodeIDs) > 0 {
q := util.MapStr{
"query": util.MapStr{
"bool": util.MapStr{
"must": []util.MapStr{
{
"terms": util.MapStr{
"id": nodeIDs,
},
},
{
"term": util.MapStr{
"agent_id": util.MapStr{
"value": id,
},
},
},
},
},
},
}
err = orm.DeleteBy(model.ESNodeInfo{}, util.MustToJSONBytes(q))
if err != nil {
log.Error(err)
h.WriteError(w, err.Error(), http.StatusInternalServerError)
return
}
q = util.MapStr{
"query": util.MapStr{
"bool": util.MapStr{
"must": []util.MapStr{
{
"terms": util.MapStr{
"metadata.labels.node_uuid": nodeIDs,
},
},
{
"term": util.MapStr{
"metadata.labels.agent_id": util.MapStr{
"value": id,
},
},
},
},
},
},
}
}
h.WriteAckOKJSON(w)
}
//
//func (h *APIHandler) tryConnect(w http.ResponseWriter, req *http.Request, ps httprouter.Params) {
// var reqBody = struct {
// Endpoint string `json:"endpoint"`
// BasicAuth model.BasicAuth
// }{}
// err := h.DecodeJSON(req, &reqBody)
// if err != nil {
// h.WriteError(w, err.Error(), http.StatusInternalServerError)
// return
// }
// connectRes, err := client.GetClient().GetInstanceBasicInfo(context.Background(), reqBody.Endpoint)
// if err != nil {
// h.WriteError(w, err.Error(), http.StatusInternalServerError)
// return
// }
// h.WriteJSON(w, connectRes, http.StatusOK)
//}
func pickAgentSettings(settings []model.Setting, nodeInfo model.ESNodeInfo) *model.Setting {
for _, setting := range settings {
if setting.Metadata.Labels["node_uuid"] == nodeInfo.NodeUUID {
return &setting
}
}
return nil
}
func getAgentTaskSetting(agentID string, v model.ESNodeInfo) (*model.Setting, error) {
taskSetting, err := getSettingsByClusterID(v.ClusterID)
if err != nil {
return nil, err
}
taskSetting.Logs = &model3.LogsTask{
Enabled: true,
LogsPath: v.Path.Logs,
}
return &model.Setting{
Metadata: model.Metadata{
Category: "agent",
Name: "task",
Labels: util.MapStr{
"agent_id": agentID,
"cluster_uuid": v.ClusterUuid,
"cluster_id": v.ClusterID,
"node_uuid": v.NodeUUID,
"endpoint": fmt.Sprintf("%s://%s", v.Schema, v.PublishAddress),
},
},
Payload: util.MapStr{
"task": taskSetting,
},
}, nil
}
// getSettingsByClusterID query agent task settings with cluster id
func getSettingsByClusterID(clusterID string) (*model3.TaskSetting, error) {
err, result := querySettingsByClusterID(clusterID)
if err != nil {
return nil, err
}
setting := &model3.TaskSetting{
NodeStats: &model3.NodeStatsTask{
Enabled: true,
},
}
var (
clusterStats = true
indexStats = true
clusterHealth = true
)
keys := []string{"payload.task.cluster_stats.enabled", "payload.task.cluster_health.enabled", "payload.task.index_stats.enabled"}
for _, row := range result.Result {
if v, ok := row.(map[string]interface{}); ok {
vm := util.MapStr(v)
for _, key := range keys {
tv, _ := vm.GetValue(key)
if tv == true {
switch key {
case "payload.task.cluster_stats.enabled":
clusterStats = false
case "payload.task.index_stats.enabled":
indexStats = false
case "payload.task.cluster_health.enabled":
clusterHealth = false
}
}
}
}
}
if clusterStats {
setting.ClusterStats = &model3.ClusterStatsTask{
Enabled: true,
}
}
if indexStats {
setting.IndexStats = &model3.IndexStatsTask{
Enabled: true,
}
}
if clusterHealth {
setting.ClusterHealth = &model3.ClusterHealthTask{
Enabled: true,
}
}
return setting, nil
}
func querySettingsByClusterID(clusterID string) (error, orm.Result) {
queryDsl := util.MapStr{
"size": 500,
"query": util.MapStr{
"bool": util.MapStr{
"must": []util.MapStr{
{
"term": util.MapStr{
"metadata.labels.cluster_id": util.MapStr{
"value": clusterID,
},
},
},
},
"minimum_should_match": 1,
"should": []util.MapStr{
{
"term": util.MapStr{
"payload.task.cluster_health.enabled": util.MapStr{
"value": true,
},
},
},
{
"term": util.MapStr{
"payload.task.cluster_stats.enabled": util.MapStr{
"value": true,
},
},
},
{
"term": util.MapStr{
"payload.task.index_stats.enabled": util.MapStr{
"value": true,
},
},
},
},
},
},
}
q := orm.Query{
RawQuery: util.MustToJSONBytes(queryDsl),
}
return orm.Search(model.Setting{}, &q)
}
func GetHostInfo(ctx context.Context, agentBaseURL string) (*host.HostInfo, error) {
req := &util.Request{
Method: http.MethodGet,
Path: "/agent/host/_basic",
Context: ctx,
}
resBody := struct {
Success bool `json:"success"`
Error string `json:"error"`
HostInfo *host.HostInfo `json:"result"`
}{}
req.Body = util.MustToJSONBytes(resBody)
err := DoRequest(req, &resBody)
if err != nil {
return nil, err
}
if resBody.Success != true {
return nil, fmt.Errorf("enroll error from client: %v", resBody.Error)
}
return resBody.HostInfo, nil
}
func GetElasticProcess(ctx context.Context, agentBaseURL string, agentID string) (interface{}, error) {
req := &util.Request{
Method: http.MethodGet,
Path: fmt.Sprintf("/elasticsearch/%s/process/_elastic", agentID),
Context: ctx,
}
resBody := map[string]interface{}{}
err := DoRequest(req, &resBody)
if err != nil {
return nil, err
}
if resBody["success"] != true {
return nil, fmt.Errorf("discover host callback error: %v", resBody["error"])
}
return resBody["elastic_process"], nil
}
func GetInstanceBasicInfo(ctx context.Context, agentBaseURL string) (*model.Instance, error) {
req := &util.Request{
Method: http.MethodGet,
Path: "/_info",
Context: ctx,
}
resBody := &model.Instance{}
err := DoRequest(req, &resBody)
return resBody, err
}
func RegisterElasticsearch(ctx context.Context, agentBaseURL string, cfgs []elastic.ElasticsearchConfig) error {
reqBody, err := util.ToJSONBytes(cfgs)
if err != nil {
return err
}
req := &util.Request{
Method: http.MethodPost,
Path: "/elasticsearch/_register",
Context: ctx,
Body: reqBody,
}
resBody := util.MapStr{}
err = DoRequest(req, &resBody)
if err != nil {
return err
}
if resBody["acknowledged"] != true {
return fmt.Errorf("%v", resBody["error"])
}
return nil
}
func CreatePipeline(ctx context.Context, agentBaseURL string, body []byte) error {
req := &util.Request{
Method: http.MethodPost,
Path: "/pipeline/tasks/",
Body: body,
Context: ctx,
}
resBody := util.MapStr{}
return DoRequest(req, &resBody)
}
func DeletePipeline(ctx context.Context, agentBaseURL, pipelineID string) error {
req := &util.Request{
Method: http.MethodDelete,
Path: fmt.Sprintf("/pipeline/task/%s", pipelineID),
Context: ctx,
}
return DoRequest(req, nil)
}
func DoRequest(req *util.Request, obj interface{}) error {
panic("implement me")
}
var mTLSClient *http.Client //TODO get mTLSClient
var initOnce = sync.Once{}
func doRequest(instance *model.Instance, req *util.Request, obj interface{}) error {
var err error
var res *util.Result
initOnce.Do(func() {
if global.Env().SystemConfig.Configs.TLSConfig.TLSEnabled && global.Env().SystemConfig.Configs.TLSConfig.TLSCAFile != "" {
//init client
hClient, err := util.NewMTLSClient(
global.Env().SystemConfig.Configs.TLSConfig.TLSCAFile,
global.Env().SystemConfig.Configs.TLSConfig.TLSCertFile,
global.Env().SystemConfig.Configs.TLSConfig.TLSKeyFile)
if err != nil {
panic(err)
}
mTLSClient = hClient
}
})
req.Url, err = url.JoinPath(instance.GetEndpoint(), req.Path)
res, err = util.ExecuteRequestWithCatchFlag(mTLSClient, req, true)
if err != nil || res.StatusCode != 200 {
body := ""
if res != nil {
body = string(res.Body)
}
return errors.New(fmt.Sprintf("request error: %v, %v", err, body))
}
if res != nil {
if res.Body != nil {
return util.FromJSONBytes(res.Body, obj)
}
}
return nil
}

455
modules/agent/common/helper.go
View File

@ -1,455 +0,0 @@
/* Copyright © INFINI Ltd. All rights reserved.
* Web: https://infinilabs.com
* Email: hello#infini.ltd */
package common
import (
"fmt"
log "github.com/cihub/seelog"
model2 "infini.sh/console/modules/agent/model"
"infini.sh/framework/core/credential"
"infini.sh/framework/core/elastic"
"infini.sh/framework/core/event"
"infini.sh/framework/core/global"
"infini.sh/framework/core/model"
"infini.sh/framework/core/orm"
"infini.sh/framework/core/util"
"strings"
)
func ParseAgentSettings(settings []model.Setting)(*model2.ParseAgentSettingsResult, error){
var clusterCfgs []elastic.ElasticsearchConfig
var (
pipelines []util.MapStr
toDeletePipelineNames []string
)
for _, setting := range settings {
if setting.Metadata.Labels == nil {
return nil, fmt.Errorf("empty metadata labels of setting [%s]", setting.ID)
}
var (
clusterID string
ok bool
)
nodeUUID := util.ToString(setting.Metadata.Labels["node_uuid"])
if clusterID, ok = setting.Metadata.Labels["cluster_id"].(string); ok && clusterID != ""{
cfg := elastic.GetConfig(clusterID)
newID := getClusterConfigReferenceName(clusterID, nodeUUID)
newCfg := elastic.ElasticsearchConfig{
Enabled: true,
Name: newID,
BasicAuth: cfg.BasicAuth,
//todo get endpoint from agent node info
Endpoint: setting.Metadata.Labels["endpoint"].(string),
ClusterUUID: cfg.ClusterUUID,
}
newCfg.ID = newID
clusterCfgs = append(clusterCfgs, newCfg)
}else{
return nil, fmt.Errorf("got wrong cluster id [%v] from metadata labels", setting.Metadata.Labels["cluster_id"])
}
taskCfg, err := util.MapStr(setting.Payload).GetValue("task")
if err != nil {
return nil, err
}
vBytes, err := util.ToJSONBytes(taskCfg)
if err != nil {
return nil, err
}
taskSetting := model2.TaskSetting{}
err = util.FromJSONBytes(vBytes, &taskSetting)
if err != nil {
return nil, err
}
partPipelines, partDeletePipelineNames, err := TransformSettingsToConfig(&taskSetting, clusterID, nodeUUID)
if err != nil {
return nil, err
}
pipelines = append(pipelines, partPipelines...)
toDeletePipelineNames = append(toDeletePipelineNames, partDeletePipelineNames...)
}
return &model2.ParseAgentSettingsResult{
ClusterConfigs: clusterCfgs,
Pipelines: pipelines,
ToDeletePipelineNames: toDeletePipelineNames,
}, nil
}
// GetAgentSettings query agent setting by agent id and updated timestamp,
// if there has any setting was updated, then return setting list includes settings not changed,
// otherwise return empty setting list
func GetAgentSettings(agentID string, timestamp int64) ([]model.Setting, error) {
query := util.MapStr{
"bool": util.MapStr{
"must": []util.MapStr{
{
"term": util.MapStr{
"metadata.category": util.MapStr{
"value": "agent",
},
},
},
{
"term": util.MapStr{
"metadata.name": util.MapStr{
"value": "task",
},
},
},
{
"term": util.MapStr{
"metadata.labels.agent_id": util.MapStr{
"value": agentID,
},
},
},
//{
// "range": util.MapStr{
// "updated": util.MapStr{
// "gt": timestamp,
// },
// },
//},
},
},
}
queryDsl := util.MapStr{
"size": 1000,
"query": query,
}
q := orm.Query{
RawQuery: util.MustToJSONBytes(queryDsl),
}
err, result := orm.Search(model.Setting{}, &q)
if err != nil {
return nil, fmt.Errorf("search settings error: %w", err)
}
if len(result.Result) == 0 {
return nil, nil
}
var (
settings []model.Setting
hasUpdated bool
)
for _, row := range result.Result {
setting := model.Setting{}
buf, err := util.ToJSONBytes(row)
if err != nil {
return nil, err
}
err = util.FromJSONBytes(buf, &setting)
if err != nil {
return nil, err
}
if setting.Updated != nil && setting.Updated.UnixMilli() > timestamp {
hasUpdated = true
}
settings = append(settings, setting)
}
if !hasUpdated {
return nil, nil
}
return settings, nil
}
func getClusterConfigReferenceName(clusterID, nodeUUID string) string {
return fmt.Sprintf("%s_%s", clusterID, nodeUUID)
}
func TransformSettingsToConfig(setting *model2.TaskSetting, clusterID, nodeUUID string) ([]util.MapStr, []string, error) {
if setting == nil {
return nil, nil, fmt.Errorf("empty setting")
}
var (
pipelines []util.MapStr
toDeletePipelineNames []string
)
if setting.ClusterStats != nil {
var processorName = "es_cluster_stats"
if setting.ClusterStats.Enabled {
pipelineCfg, err := newClusterMetricPipeline(processorName, clusterID, nodeUUID)
if err != nil {
return nil, nil, err
}
pipelines = append(pipelines, pipelineCfg)
}else{
toDeletePipelineNames = append(toDeletePipelineNames, getMetricPipelineName(clusterID, processorName))
}
}
if setting.IndexStats != nil {
var processorName = "es_index_stats"
if setting.IndexStats.Enabled {
pipelineCfg, err := newClusterMetricPipeline(processorName, clusterID, nodeUUID)
if err != nil {
return nil, nil, err
}
pipelines = append(pipelines, pipelineCfg)
}else{
toDeletePipelineNames = append(toDeletePipelineNames, getMetricPipelineName(clusterID, processorName))
}
}
if setting.ClusterHealth != nil {
var processorName = "es_cluster_health"
if setting.ClusterHealth.Enabled {
pipelineCfg, err := newClusterMetricPipeline(processorName, clusterID, nodeUUID)
if err != nil {
return nil, nil, err
}
pipelines = append(pipelines, pipelineCfg)
}else{
toDeletePipelineNames = append(toDeletePipelineNames, getMetricPipelineName(clusterID, processorName))
}
}
if setting.NodeStats != nil {
var processorName = "es_node_stats"
if setting.NodeStats.Enabled {
params := util.MapStr{
"elasticsearch": getClusterConfigReferenceName(clusterID, nodeUUID),
"labels": util.MapStr{
"cluster_id": clusterID,
},
}
if len(setting.NodeStats.NodeIDs) > 0{
params["node_uuids"] = setting.NodeStats.NodeIDs
}
cfg := util.MapStr{
processorName: params,
}
enabled := true
pipelineCfg := util.MapStr{
"enabled": &enabled,
"name": getMetricPipelineName(nodeUUID, processorName),
"auto_start": true,
"keep_running": true,
"retry_delay_in_ms": 10000,
"processor": []util.MapStr{cfg},
}
pipelines = append(pipelines, pipelineCfg)
}else{
toDeletePipelineNames = append(toDeletePipelineNames, getMetricPipelineName(nodeUUID, processorName))
}
}
if setting.Logs != nil {
var processorName = "es_logs_processor"
if setting.Logs.Enabled {
params := util.MapStr{
"elasticsearch": getClusterConfigReferenceName(clusterID, nodeUUID),
"queue_name": "logs",
"labels": util.MapStr{
"cluster_id": clusterID,
},
}
if setting.Logs.LogsPath != "" {
params["logs_path"] = setting.Logs.LogsPath
}
cfg := util.MapStr{
processorName: params,
}
enabled := true
pipelineCfg := util.MapStr{
"enabled": &enabled,
"name": fmt.Sprintf("collect_%s_es_logs", nodeUUID),
"auto_start": true,
"keep_running": true,
"retry_delay_in_ms": 3000,
"processor": []util.MapStr{cfg},
}
pipelines = append(pipelines, pipelineCfg)
}
}
return pipelines, toDeletePipelineNames, nil
}
func newClusterMetricPipeline(processorName string, clusterID string, nodeUUID string)(util.MapStr, error){
referName := getClusterConfigReferenceName(clusterID, nodeUUID)
cfg := util.MapStr{
processorName: util.MapStr{
"elasticsearch": referName,
"labels": util.MapStr{
"cluster_id": clusterID,
},
},
}
enabled := true
pipelineCfg := util.MapStr{
"enabled": &enabled,
"name": getMetricPipelineName(clusterID, processorName),
"auto_start": true,
"keep_running": true,
"singleton": true,
"retry_delay_in_ms": 10000,
"processor": []util.MapStr{cfg},
}
return pipelineCfg, nil
}
func getMetricPipelineName(clusterID, processorName string) string{
return fmt.Sprintf("collect_%s_%s", clusterID, processorName)
}
func LoadAgentsFromES(clusterID string) ([]model.Instance, error) {
q := orm.Query{
Size: 1000,
}
if clusterID != "" {
q.Conds = orm.And(orm.Eq("id", clusterID))
}
err, result := orm.Search(model.Instance{}, &q)
if err != nil {
return nil, fmt.Errorf("query agent error: %w", err)
}
if len(result.Result) > 0 {
var agents = make([]model.Instance, 0, len(result.Result))
for _, row := range result.Result {
ag := model.Instance{}
bytes := util.MustToJSONBytes(row)
err = util.FromJSONBytes(bytes, &ag)
if err != nil {
log.Errorf("got unexpected agent: %s, error: %v", string(bytes), err)
continue
}
agents = append(agents, ag)
}
return agents, nil
}
return nil, nil
}
func GetLatestOnlineAgentIDs(agentIds []string, lastSeconds int) (map[string]struct{}, error) {
q := orm.Query{
WildcardIndex: true,
}
mustQ := []util.MapStr{
{
"term": util.MapStr{
"metadata.name": util.MapStr{
"value": "agent",
},
},
},
{
"term": util.MapStr{
"metadata.category": util.MapStr{
"value": "instance",
},
},
},
}
if len(agentIds) > 0 {
mustQ = append(mustQ, util.MapStr{
"terms": util.MapStr{
"agent.id": agentIds,
},
})
}
queryDSL := util.MapStr{
"sort": []util.MapStr{
{
"timestamp": util.MapStr{
"order": "desc",
},
},
},
"collapse": util.MapStr{
"field": "agent.id",
},
"query": util.MapStr{
"bool": util.MapStr{
"filter": []util.MapStr{
{
"range": util.MapStr{
"timestamp": util.MapStr{
"gte": fmt.Sprintf("now-%ds", lastSeconds),
},
},
},
},
"must": mustQ,
},
},
}
if len(agentIds) == 0 {
queryDSL["size"] = 2000
}else{
queryDSL["size"] = len(agentIds)
}
q.RawQuery = util.MustToJSONBytes(queryDSL)
err, result := orm.Search(event.Event{}, &q)
if err != nil {
return nil, fmt.Errorf("query agent instance metric error: %w", err)
}
agentIDs := map[string]struct{}{}
if len(result.Result) > 0 {
searchRes := elastic.SearchResponse{}
err = util.FromJSONBytes(result.Raw, &searchRes)
if err != nil {
return nil, err
}
agentIDKeyPath := []string{"agent", "id"}
for _, hit := range searchRes.Hits.Hits {
agentID, _ := util.GetMapValueByKeys(agentIDKeyPath, hit.Source)
if v, ok := agentID.(string); ok {
agentIDs[v] = struct{}{}
}
}
}
return agentIDs, nil
}
func GetAgentIngestConfig() (string, *model.BasicAuth, error) {
agCfg := GetAgentConfig()
var (
endpoint string
ok bool
)
emptyIngestClusterEndpoint := false
if agCfg.Setup.IngestClusterEndpoint == nil {
emptyIngestClusterEndpoint = true
}
if endpoint, ok = agCfg.Setup.IngestClusterEndpoint.(string);ok {
if endpoint = strings.TrimSpace(endpoint); endpoint == "" {
emptyIngestClusterEndpoint = true
}
}
if emptyIngestClusterEndpoint {
cfg := elastic.GetConfig(global.MustLookupString(elastic.GlobalSystemElasticsearchID))
endpoint = cfg.GetAnyEndpoint()
}
var (
basicAuth model.BasicAuth
)
if agCfg.Setup.IngestClusterCredentialID != "" {
cred := credential.Credential{}
cred.ID = agCfg.Setup.IngestClusterCredentialID
_, err := orm.Get(&cred)
if err != nil {
return "", nil, fmt.Errorf("query credential [%s] error: %w", cred.ID, err)
}
info, err := cred.Decode()
if err != nil {
return "", nil, fmt.Errorf("decode credential [%s] error: %w", cred.ID, err)
}
if basicAuth, ok = info.(model.BasicAuth); !ok {
log.Debug("invalid credential: ", cred)
}
}else{
cfg := elastic.GetConfig(global.MustLookupString(elastic.GlobalSystemElasticsearchID))
basicAuth = *cfg.BasicAuth
}
tpl := `configs.template:
- name: "ingest"
path: ./config/ingest_config.tpl
variable:
INGEST_CLUSTER_ID: "default_ingest_cluster"
INGEST_CLUSTER_ENDPOINT: ["%s"]
INGEST_CLUSTER_USERNAME: "%s"
`
tpl = fmt.Sprintf(tpl, endpoint, basicAuth.Username)
return tpl, &basicAuth, nil
}

5
modules/agent/model/config.go
View File

@ -6,9 +6,6 @@ package model
type AgentConfig struct {
Enabled bool `config:"enabled"`
StateManager struct {
Enabled bool `config:"enabled"`
} `config:"state_manager"`
Setup *SetupConfig `config:"setup"`
}
@ -18,7 +15,5 @@ type SetupConfig struct {
CACertFile string `config:"ca_cert"`
CAKeyFile string `config:"ca_key"`
ConsoleEndpoint string `config:"console_endpoint"`
IngestClusterEndpoint interface{} `config:"ingest_cluster_endpoint"`
IngestClusterCredentialID string `config:"ingest_cluster_credential_id"`
Port string `config:"port"`
}

46
modules/agent/model/task.go
View File

@ -1,46 +0,0 @@
/* Copyright © INFINI Ltd. All rights reserved.
* Web: https://infinilabs.com
* Email: hello#infini.ltd */
package model
import (
"infini.sh/framework/core/elastic"
"infini.sh/framework/core/util"
)
type TaskSetting struct {
ClusterHealth *ClusterHealthTask `json:"cluster_health,omitempty"`
ClusterStats *ClusterStatsTask `json:"cluster_stats,omitempty"`
IndexStats *IndexStatsTask `json:"index_stats,omitempty"`
NodeStats *NodeStatsTask `json:"node_stats,omitempty"`
Logs *LogsTask `json:"logs,omitempty"`
}
type ClusterHealthTask struct {
Enabled bool `json:"enabled"`
}
type ClusterStatsTask struct {
Enabled bool `json:"enabled"`
}
type IndexStatsTask struct {
Enabled bool `json:"enabled"`
}
type NodeStatsTask struct {
Enabled bool `json:"enabled"`
NodeIDs []string `json:"node_ids,omitempty"`
}
type LogsTask struct {
Enabled bool `json:"enabled"`
LogsPath string `json:"logs_path"`
}
type ParseAgentSettingsResult struct {
ClusterConfigs []elastic.ElasticsearchConfig
Pipelines []util.MapStr
ToDeletePipelineNames []string
}