feat: Add health checking feature to service registry, which periodically checks service status and deletes unhealthy services

internal/general-node/node.go

@@ -17,6 +17,9 @@ func Start() {
 	// Migrate the schema
 	db.AutoMigrate(&models.Service{})
 
+	// Health Checking
+	go checkHealth()
+
 	// Init API Backend
 	router := gin.Default()
 	router = registerRoutes(router)

internal/general-node/service_registry.go

@@ -2,6 +2,8 @@ package generalnode
 
 import (
 	"fmt"
+	"net/http"
+	"time"
 
 	"github.com/alexohneander/flotte/pkg/models"
 	"github.com/alexohneander/flotte/pkg/types/request"
@@ -16,7 +18,43 @@ func registerSerivce(req request.ServiceRegister) error {
 		return err
 	}
 
-	db.Create(&models.Service{Name: req.Name, NodeType: req.NodeType, Address: req.Address, Port: req.Port, Status: "running"})
+	err = db.Create(&models.Service{Name: req.Name, NodeType: req.NodeType, Address: req.Address, Port: req.Port, Status: "running"}).Error
+	if err != nil {
+		fmt.Println("failed to create service")
+		return err
+	}
 
 	return nil
 }
+
+func checkHealth() {
+	db, err := gorm.Open(sqlite.Open("flotte.db"), &gorm.Config{})
+	if err != nil {
+		fmt.Println("failed to connect database")
+	}
+
+	for {
+		var services []models.Service
+		db.Find(&services, models.Service{})
+		for _, service := range services {
+			resp, err := http.Get("http://" + service.Address + ":" + service.Port + "/ping")
+			if err != nil {
+				fmt.Println("error: ", err)
+				fmt.Println("delete service ", service.Name)
+				db.Unscoped().Delete(&service, service.ID)
+			}
+
+			if resp != nil && resp.StatusCode != 200 {
+				fmt.Println("delete service ", service.Name)
+				db.Unscoped().Delete(&service, service.ID)
+			}
+
+			if resp != nil && resp.StatusCode == 200 {
+				fmt.Println("service ", service.Name, " is healthy")
+			}
+		}
+
+		time.Sleep(30 * time.Second)
+	}
+
+}

internal/map-node/node.go

@@ -29,7 +29,7 @@ func registerAsMapNode() error {
 		Name:     "map-01",
 		NodeType: "map",
 		Address:  "localhost",
-		Port:     4000,
+		Port:     "4000",
 	}
 
 	json_data, err := json.Marshal(serviceRegisterReq)
@@ -46,6 +46,11 @@ func registerAsMapNode() error {
 		return err
 	}
 
+	if resp.StatusCode != http.StatusOK {
+		log.Fatal("failed to register as map-node")
+		return err
+	}
+
 	defer resp.Body.Close()
 	var res map[string]interface{}
 

internal/reduce-node/node.go

@@ -29,7 +29,7 @@ func registerAsReduceNode() error {
 		Name:     "reduce-01",
 		NodeType: "reduce",
 		Address:  "localhost",
-		Port:     4001,
+		Port:     "4001",
 	}
 
 	json_data, err := json.Marshal(serviceRegisterReq)
@@ -46,6 +46,11 @@ func registerAsReduceNode() error {
 		return err
 	}
 
+	if resp.StatusCode != http.StatusOK {
+		log.Fatal("failed to register as reduce-node")
+		return err
+	}
+
 	defer resp.Body.Close()
 	var res map[string]interface{}
 

pkg/models/service.go

@@ -4,9 +4,9 @@ import "gorm.io/gorm"
 
 type Service struct {
 	gorm.Model
-	Name     string
+	Name     string `gorm:"unique"`
 	NodeType string
 	Address  string
-	Port     int
+	Port     string
 	Status   string
 }

pkg/types/request/service_register.go

@@ -4,5 +4,5 @@ type ServiceRegister struct {
 	Name     string `json:"name"`
 	NodeType string `json:"nodeType"`
 	Address  string `json:"address"`
-	Port     int    `json:"port"`
+	Port     string `json:"port"`
 }