dragonheim/gagent
dragonheim/gagent
1
0
Fork 0
mirror of https://github.com/dragonheim/gagent.git synced 2025-04-25 22:48:59 -07:00
Code Issues Projects Releases Packages Wiki Activity Actions

Extending prometheus support.

Browse source
This commit is contained in:
James Wells 2021-10-18 05:46:52 -07:00
parent 16d40fd93a
commit 8a7842e6bc
Signed by: jwells
GPG key ID: 73196D10B8E65666
4 changed files with 62 additions and 37 deletions
Download patch file Download diff file Expand all files Collapse all files

8
internal/client/client.go
View file

@ -12,7 +12,13 @@ import (
zmq "github.com/pebbe/zmq4"
)
// Main is the initiation function for a Client
/*
Client mode will send an agent file to a router for processing
Clients do not process the agent files, only send them as
requests to a router. If started without arguments, the client
will contact the router and attempt to retrieve the results
of it's most recent request.
*/
func Main(wg *sync.WaitGroup, config gs.GagentConfig, rid int, agent string) {
defer wg.Done()
log.Printf("[INFO] Starting client\n")

30
internal/router/router.go
View file

@ -8,7 +8,8 @@ import (
gs "git.dragonheim.net/dragonheim/gagent/internal/gstructs"
promhttp "github.com/prometheus/client_golang/prometheus/promhttp"
prometheus "github.com/prometheus/client_golang/prometheus"
promauto "github.com/prometheus/client_golang/prometheus/promauto"
zmq "github.com/pebbe/zmq4"
)
@ -18,10 +19,21 @@ const (
WORKER_READY = "\001" // Signals worker is ready
)
// Main is the initiation function for a Router
var (
opsProcessed = promauto.NewCounter(prometheus.CounterOpts{
Name: "client_requests_recieved",
})
)
/*
The 'router' processes routing requests from the agent. The router does
not handle any of the agent activities beyond processing the agent's
list of tags and passing the agent and it's storage to either a member
or client node. Tags are used by the agent to give hints as to where
it should be routed.
*/
func Main(wg *sync.WaitGroup, config gs.GagentConfig) {
defer wg.Done()
http.Handle("/metrics", promhttp.Handler())
http.HandleFunc("/hello", answerClient)
log.Printf("[INFO] Starting router\n")
@ -31,13 +43,8 @@ func Main(wg *sync.WaitGroup, config gs.GagentConfig) {
workerSock, _ := zmq.NewSocket(zmq.DEALER)
defer workerSock.Close()
workerListener := fmt.Sprintf("tcp://%s:%d", config.ListenAddr, config.WorkerPort)
clientListener := fmt.Sprintf("%s:%d", config.ListenAddr, config.ClientPort)
go func() {
http.ListenAndServe(clientListener, nil)
}()
workerSock.Bind(workerListener)
workers := make([]string, 0)
@ -106,8 +113,8 @@ func unwrap(msg []string) (head string, tail []string) {
func answerClient(w http.ResponseWriter, r *http.Request) {
if r.URL.Path != "/" {
fmt.Fprintf(w, "%v\n", r)
// http.NotFound(w, r)
// fmt.Fprintf(w, "%v\n", r)
http.NotFound(w, r)
return
}
@ -115,6 +122,7 @@ func answerClient(w http.ResponseWriter, r *http.Request) {
switch r.Method {
case http.MethodGet:
opsProcessed.Inc()
fmt.Fprintf(w, "%v\n", r)
// Handle the GET request...

18
internal/worker/worker.go
View file

@ -9,26 +9,38 @@ import (
gs "git.dragonheim.net/dragonheim/gagent/internal/gstructs"
// picol "git.dragonheim.net/dragonheim/gagent/src/picol"
promhttp "github.com/prometheus/client_golang/prometheus/promhttp"
zmq "github.com/pebbe/zmq4"
)
// Main is the initiation function for a Worker
/*
The "worker" processes the agent code. The worker nodes do not know
anything about the network structure. Instead they know only to which
router(s) they are connected. The worker will execute the agent code and
pass the agent and it's results to a router.
*/
func Main(wg *sync.WaitGroup, config gs.GagentConfig, rid int) {
defer wg.Done()
http.Handle("/metrics", promhttp.Handler())
log.Printf("[INFO] Starting worker\n")
// Generate connect string for this router.
var rport = int64(config.WorkerPort)
var rport = config.WorkerPort
if config.Routers[rid].WorkerPort != 0 {
rport = config.Routers[rid].WorkerPort
}
connectString := fmt.Sprintf("tcp://%s:%d", config.Routers[rid].RouterAddr, rport)
// workerListener := fmt.Sprintf("tcp://%s:%d", config.ListenAddr, config.WorkerPort)
clientListener := fmt.Sprintf("%s:%d", config.ListenAddr, config.ClientPort)
subscriber, _ := zmq.NewSocket(zmq.REP)
defer subscriber.Close()
go func() {
http.ListenAndServe(fmt.Sprintf("%s:%d", config.ListenAddr, config.ClientPort), nil)
http.ListenAndServe(clientListener, nil)
}()
log.Printf("[DEBUG] Attempting to connect to %s\n", connectString)