group requests to reduce agent calls

Author: muraty

agent/agent.go

@@ -14,7 +14,7 @@ import (
 // Agent is type for handling agent operations
 type Agent struct {
 	Config   *config.AgentConfig
-	conns    []gopsutilNet.ConnectionStat
+	conns    []gopsutilNet.ConnectionStat //todo: store only what we need.
 	Hostname string
 	mu       sync.RWMutex // protects conns
 }

agent/http.go

@@ -2,112 +2,127 @@ package agent
 
 import (
 	"encoding/json"
+	"fmt"
 	"net/http"
 	"strconv"
+	"strings"
 
 	"github.com/cenkalti/log"
 	gopsutilNet "github.com/shirou/gopsutil/v4/net"
 	gopsutilProcess "github.com/shirou/gopsutil/v4/process"
 )
 
-// Response contains basic process information for API responses.
-type Response struct {
+// Process contains basic process information for API responses.
+type Process struct {
 	Status  string `json:"status"`
 	Pid     int32  `json:"pid"`
+	Port    int32  `json:"port"`
 	Name    string `json:"name"`
 	CmdLine string `json:"cmdline"`
 }
 
-// NetworkProcess represents process with its network connection.
-type NetworkProcess struct {
-	process *gopsutilProcess.Process
-	conn    gopsutilNet.ConnectionStat
+// Response contains basic process information for API responses.
+type Response struct {
+	Processes []*Process `json:"processes"`
 }
 
-// parsePortParam parses and returns port number from the request.
-func parsePortParam(w http.ResponseWriter, req *http.Request) (uint32, error) {
-	portParam, ok := req.URL.Query()["port"]
-	log.Debugf("Looking for process of port: %s\n", portParam)
+// parsePortsParam parses and returns port numbers from the request.
+func parsePortsParam(w http.ResponseWriter, req *http.Request) ([]uint32, error) {
+	portsParam := req.URL.Query().Get("ports")
+	log.Debugf("Looking for process(es) for ports: %s\n", portsParam)
 
-	if !ok || len(portParam) < 1 {
-		log.Errorln("port param is not provided.")
-		return 0, nil
+	if portsParam == "" {
+		log.Errorln("ports param is not provided.")
+		return nil, fmt.Errorf("ports param is required") //todo: check again.
 	}
 
-	p, err := strconv.ParseInt(portParam[0], 10, 32)
-	if err != nil {
-		log.Errorln("error during string to int32: %s\n", err)
-		return 0, err
+	ports := strings.Split(portsParam, ",")
+
+	var portNumbers []uint32
+	for _, port := range ports {
+		p, err := strconv.Atoi(port)
+		if err != nil {
+			return nil, fmt.Errorf("invalid port number: %s", port)
+		}
+		portNumbers = append(portNumbers, uint32(p))
 	}
-	return uint32(p), nil
+
+	return portNumbers, nil
 }
 
-// findProcess finds process from connections by given port.
-func findProcess(port uint32, conns []gopsutilNet.ConnectionStat) *NetworkProcess {
+// findProcesses finds process(es) those have connections with given ports.
+func findProcesses(ports []uint32, conns []gopsutilNet.ConnectionStat) []*Process {
+	ps := make([]*Process, 0)
+
 	for _, conn := range conns {
-		if conn.Laddr.Port != port {
+		if !portExists(conn.Laddr.Port, ports) {
 			continue
 		}
 
 		process, err := gopsutilProcess.NewProcess(conn.Pid)
 		if err != nil {
 			log.Debugf("Error occured while finding the process %s\n", err.Error())
-			return nil
+			continue
 		}
 
 		if process == nil {
-			log.Debugf("Process could not found for %d\n", conn.Pid)
-			return nil
+			log.Debugf("Process not found for %d\n", conn.Pid)
+			continue
+		}
+
+		name, err := process.Name()
+		if err != nil {
+			log.Debugf("Name not found for %d\n", process.Pid)
 		}
 
-		return &NetworkProcess{
-			process: process,
-			conn:    conn,
+		cmdline, err := process.Cmdline()
+		if err != nil {
+			log.Debugf("Cmdline not found for %d\n", process.Pid)
 		}
+
+		p := &Process{
+			Status:  conn.Status,
+			Pid:     conn.Pid,
+			Port:    int32(conn.Laddr.Port),
+			Name:    name,
+			CmdLine: cmdline,
+		}
+
+		ps = append(ps, p)
 	}
-	return nil
+	return ps
 }
 
-// createResponse creates Response from given NetworkProcess parameter.
-func createResponse(np *NetworkProcess) *Response {
-	if np == nil {
-		return nil
-	}
-	name, err := np.process.Name()
-	if err != nil {
-		name = ""
-	}
-	cmdline, err := np.process.Cmdline()
-	if err != nil {
-		log.Debugf("Cmdline could not found for %d\n", np.process.Pid)
-	}
-	return &Response{
-		Status:  np.conn.Status,
-		Pid:     np.conn.Pid,
-		Name:    name,
-		CmdLine: cmdline,
+func portExists(port uint32, ports []uint32) bool {
+	for _, p := range ports {
+		if p == port {
+			return true
+		}
 	}
+	return false
 }
 
 // Process is handler for serving process info
 func (a *Agent) Process(w http.ResponseWriter, req *http.Request) {
 	w.Header().Set("X-Kimo-Hostname", a.Hostname)
 
 	// todo: cache result for a short period (10s? 30s?)
-	port, err := parsePortParam(w, req)
+	ports, err := parsePortsParam(w, req)
 	if err != nil {
 		http.Error(w, "port param is required", http.StatusBadRequest)
 		return
 	}
-	p := findProcess(port, a.GetConns())
-	if p == nil {
-		http.Error(w, "Connection not found", http.StatusNotFound)
+	ps := findProcesses(ports, a.GetConns())
+	if ps == nil {
+		http.Error(w, "Connection(s) not found", http.StatusNotFound)
 		return
 	}
 
 	w.Header().Set("Content-Type", "application/json")
-	ap := createResponse(p)
-	err = json.NewEncoder(w).Encode(&ap)
+	response := &Response{
+		Processes: ps,
+	}
+	err = json.NewEncoder(w).Encode(&response)
 	if err != nil {
 		http.Error(w, "Can not encode agent process", http.StatusInternalServerError)
 	}

kimo-agent-entrypoint.sh

@@ -19,8 +19,10 @@ if [ "$i" = 0 ]; then
         exit 1
 fi
 
-echo "mysql sleep query..."
-mysql -u"$user" -p"$password" -h"$host" -e "SELECT SLEEP(100000)" &
+echo "mysql sleep queries..."
+mysql -u"$user" -p"$password" -h"$host" -e "SELECT SLEEP(5000)" &
+mysql -u"$user" -p"$password" -h"$host" -e "SELECT SLEEP(6000)" &
+mysql -u"$user" -p"$password" -h"$host" -e "SELECT SLEEP(7000)" &
 
 echo "running kimo agent..."
 kimo --debug agent

server/agent.go

@@ -5,13 +5,14 @@ import (
 	"encoding/json"
 	"fmt"
 	"net/http"
+	"strings"
 
 	"github.com/cenkalti/log"
 )
 
 // AgentProcess represents process info from a kimo-agent enhanced with response detail.
 type AgentProcess struct {
-	Pid              int
+	Pid              uint32
 	Name             string
 	Cmdline          string
 	ConnectionStatus string
@@ -31,6 +32,7 @@ func NewAgentProcesss(ar *AgentResponse, address IPPort, err error) *AgentProces
 		ap.Name = ar.Name
 		ap.Cmdline = ar.CmdLine
 		ap.hostname = ar.Hostname
+		ap.ConnectionStatus = ar.Status
 	}
 	return ap
 }
@@ -50,8 +52,7 @@ func (ap *AgentProcess) Host() string {
 
 // AgentClient represents an agent client to fetch get process from a kimo-agent
 type AgentClient struct {
-	Host string
-	Port uint32
+	Address IPPort
 }
 
 // AgentError represents an HTTP error that is retured from kimo-agent.
@@ -62,11 +63,12 @@ type AgentError struct {
 
 // AgentResponse represents a success response from kimo-agent.
 type AgentResponse struct {
-	Pid              int
-	Name             string
-	CmdLine          string
-	Hostname         string
-	ConnectionStatus string
+	Hostname string
+	Status   string
+	Pid      uint32
+	Name     string
+	CmdLine  string
+	Port     uint32
 }
 
 func (ae *AgentError) Error() string {
@@ -75,15 +77,13 @@ func (ae *AgentError) Error() string {
 
 // NewAgentClient creates and returns a new AgentClient.
 func NewAgentClient(address IPPort) *AgentClient {
-	ac := new(AgentClient)
-	ac.Host = address.IP
-	ac.Port = address.Port
-	return ac
+	// kimo-agent listens this address
+	return &AgentClient{Address: address}
 }
 
 // Get gets process info from kimo agent.
-func (ac *AgentClient) Get(ctx context.Context, port uint32) (*AgentResponse, error) {
-	url := fmt.Sprintf("http://%s:%d/proc?port=%d", ac.Host, ac.Port, port)
+func (ac *AgentClient) Get(ctx context.Context, ports []uint32) ([]*AgentResponse, error) {
+	url := fmt.Sprintf("http://%s:%d/proc?ports=%s", ac.Address.IP, ac.Address.Port, createPortsParam(ports))
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
 	if err != nil {
 		return nil, err
@@ -101,27 +101,48 @@ func (ac *AgentClient) Get(ctx context.Context, port uint32) (*AgentResponse, er
 		return nil, &AgentError{Hostname: hostname, Status: response.Status}
 	}
 
-	type result struct {
+	type process struct {
 		Status  string `json:"status"`
-		Pid     int32  `json:"pid"`
+		Pid     uint32 `json:"pid"`
+		Port    uint32 `json:"port"`
 		Name    string `json:"name"`
 		CmdLine string `json:"cmdline"`
 	}
-	r := result{}
+
+	type Response struct {
+		Processes []*process `json:"processes"`
+	}
+
+	var r Response
 	err = json.NewDecoder(response.Body).Decode(&r)
 
 	if err != nil {
 		log.Errorln(err.Error())
 		return nil, err
 	}
 
-	return &AgentResponse{
-			ConnectionStatus: r.Status,
-			Pid:              int(r.Pid),
-			Name:             r.Name,
-			CmdLine:          r.CmdLine,
-			Hostname:         hostname},
-		nil
+	ars := make([]*AgentResponse, 0)
+	for _, p := range r.Processes {
+		ar := &AgentResponse{
+			Status:   p.Status,
+			Pid:      p.Pid,
+			Port:     p.Port,
+			Name:     p.Name,
+			CmdLine:  p.CmdLine,
+			Hostname: hostname,
+		}
+		ars = append(ars, ar)
+	}
+
+	return ars, nil
+
+}
+func createPortsParam(ports []uint32) string {
+	numbers := make([]string, len(ports))
+	for i, port := range ports {
+		numbers[i] = fmt.Sprint(port)
+	}
+	return strings.Join(numbers, ",")
 }
 
 func findAgentProcess(addr IPPort, aps []*AgentProcess) *AgentProcess {