Proof-of-Concept splitting of reader and writers.

client/Start.go

@@ -2,7 +2,6 @@ package client
 
 import (
 	"log"
-	"net/url"
 	"os"
 	"os/signal"
 	"strconv"
@@ -11,62 +10,77 @@ import (
 	"github.com/gorilla/websocket"
 	"github.com/inhies/go-bytesize"
 	"github.com/mackerelio/go-osstat/memory"
-	"github.com/shirou/gopsutil/v4/cpu"
+	"github.com/shirou/gopsutil/cpu"
 )
 
 func (client *Client) Start() {
-	// Setup the interrupts
+
+	// Setup interrupt handling
 	interrupt := make(chan os.Signal, 1)
 	signal.Notify(interrupt, os.Interrupt)
 
-	u := url.URL{Scheme: "ws", Host: client.ServerAddress + ":" + strconv.Itoa(client.ServerPort), Path: "/ws"}
-	log.Printf("Connecting to %s", u.String())
-
-	// sigh
-	var err error
-	client.conn, _, err = websocket.DefaultDialer.Dial(u.String(), nil)
+	// Connect to the server
+	err := client.connectToServer()
 	if err != nil {
-		log.Fatal("dial:", err)
+		log.Fatal("ERROR: Failed connecting to server - ", err)
 	}
 	defer client.conn.Close()
 
-	done := make(chan struct{})
+	// Agent => Manager:
+	// - register: register an agent with the manager
+	// - update: agents sends CPU and Mem metrics (does this every second)
+	// - solution: solution of a task given by the manager
+	// - deregister: deregister an agent from the manager
+
+	// Register
+	// One-time at the start
+
+	// Fetch the hostname
+	hostname, err := os.Hostname()
+	if err != nil {
+		log.Fatal("ERROR: Failed to register - cannot get hostname - ", err)
+	}
+
+	// Fetch the number of logical cores
+	cpustats, err := cpu.Counts(true)
+	if err != nil {
+		log.Fatal("ERROR: Failed to register - unable to fetch number of cores - ", err)
+	}
+
+	// Fetch memory stats in bytes
+	mem, err := memory.Get()
+	if err != nil {
+		log.Println("ERROR: Failed to register - fetching memory info - ", err)
+	}
+	// Use the ByteSize package to allow for memory calculations
+	b := bytesize.New(float64(mem.Total))
+
+	// Register this agent with the scheduler
+	client.writeToServer("register;" + hostname + ";" + strconv.Itoa(cpustats) + ";" + b.String())
 
 	// Setup updater logic
-	updater := make(chan string)
-	go client.statusUpdater(updater)
+	// Runs continuesly in the background
+	go client.statusUpdater()
 
-	go func() {
-		defer close(done)
-		for {
-			_, message, err := client.conn.ReadMessage()
-			if err != nil {
-				log.Println("read:", err)
-				return
-			}
-			log.Printf("recv: %s", message)
-		}
-	}()
+	// Start handler for incoming messages
+	done := make(chan struct{})
+	go client.handleIncoming(done)
 
 	for {
 		select {
 		case <-done:
 			return
-		case update := <-updater: // Dumping this into a var saves a second. Weird!
-			// Dump the message towards the server.
-			err = client.conn.WriteMessage(websocket.TextMessage, []byte(update))
-			if err != nil {
-				log.Println("write:", err)
-				return
-			}
 		case <-interrupt:
-			log.Println("interrupt")
+			log.Println("Interrupt received -- Stopping")
+
+			// Dereg the agent before dying off
+			client.writeToServer("deregister;" + hostname)
 
 			// Cleanly close the connection by sending a close message and then
 			// waiting (with timeout) for the server to close the connection.
 			err := client.conn.WriteMessage(websocket.CloseMessage, websocket.FormatCloseMessage(websocket.CloseNormalClosure, ""))
 			if err != nil {
-				log.Println("write close:", err)
+				log.Println("ERROR: Close connection - ", err)
 				return
 			}
 			select {
@@ -78,25 +92,3 @@ func (client *Client) Start() {
 	}
 
 }
-
-func (client *Client) statusUpdater(updater chan string) {
-	for {
-		// Fetch CPU usage in percentages
-		cpustats, err := cpu.Percent(time.Second, false)
-		if err != nil {
-			log.Println("Err:", err)
-		}
-
-		// Fetch memory stats in bytes
-		mem, err := memory.Get()
-		if err != nil {
-			log.Println("Err:", err)
-		}
-
-		// Use the ByteSize package to allow for memory calculations
-		b := bytesize.New(float64(mem.Used))
-
-		// Dump the message towards the server.
-		updater <- "update;" + strconv.Itoa(int(cpustats[0])) + ";" + b.String() + ";" + strconv.Itoa(client.taskId)
-	}
-}

client/connectToServer.go

@@ -0,0 +1,22 @@
+package client
+
+import (
+	"log"
+	"net/url"
+	"strconv"
+
+	"github.com/gorilla/websocket"
+)
+
+func (client *Client) connectToServer() (err error) {
+	u := url.URL{Scheme: "ws", Host: client.ServerAddress + ":" + strconv.Itoa(client.ServerPort), Path: "/ws"}
+	log.Printf("Connecting to %s", u.String())
+
+	// sigh
+	client.conn, _, err = websocket.DefaultDialer.Dial(u.String(), nil)
+	if err != nil {
+		log.Fatal("ERROR: dialing - :", err)
+		return err
+	}
+	return nil
+}

client/handleIncoming.go

@@ -0,0 +1,26 @@
+package client
+
+import (
+	"log"
+	"strings"
+)
+
+func (client *Client) handleIncoming(done chan struct{}) {
+	defer close(done)
+	for {
+		_, message, err := client.conn.ReadMessage()
+		if err != nil {
+			log.Println("ERROR: read message - :", err)
+			return
+		}
+		// Ignore printing keep alives
+		if string(message) == "keep alive - staying alive" {
+			continue
+		}
+		log.Printf("recv: %s", message)
+		inputs := strings.Split(string(message), ";")
+		if inputs[0] == "task" {
+			client.task(inputs)
+		}
+	}
+}

client/statusUpdater.go

@@ -0,0 +1,38 @@
+package client
+
+import (
+	"log"
+	"strconv"
+	"time"
+
+	"github.com/inhies/go-bytesize"
+	"github.com/mackerelio/go-osstat/memory"
+	"github.com/shirou/gopsutil/cpu"
+)
+
+// Look into using a ticker instead of a for-loop
+func (client *Client) statusUpdater() {
+	for {
+		// Fetch CPU usage in percentages
+		cpustats, err := cpu.Percent(time.Second, false)
+		if err != nil {
+			log.Println("ERROR: fetching CPU usage infor - ", err)
+		}
+
+		// Fetch memory stats in bytes
+		mem, err := memory.Get()
+		if err != nil {
+			log.Println("ERROR: fetching memory info - ", err)
+		}
+
+		// Use the ByteSize package to allow for memory calculations
+		b := bytesize.New(float64(mem.Used))
+
+		// Prepare the message.
+		update := "update;" + strconv.Itoa(int(cpustats[0])) + ";" + b.String() + ";" + strconv.Itoa(client.taskId)
+
+		// Write the message to the server
+		client.writeToServer(update)
+
+	}
+}

client/task.go

@@ -0,0 +1,7 @@
+package client
+
+import "log"
+
+func (client *Client) task(inputs []string) {
+	log.Println(inputs)
+}

client/writeToServer.go

@@ -0,0 +1,16 @@
+package client
+
+import (
+	"log"
+
+	"github.com/gorilla/websocket"
+)
+
+func (client *Client) writeToServer(msg string) {
+	// Send the message towards the server.
+	err := client.conn.WriteMessage(websocket.TextMessage, []byte(msg))
+	if err != nil {
+		log.Println("ERROR: writing - ", err)
+		return
+	}
+}