feat: finish litefs config env

Author: VaalaCat

docker-compose-dev.yml

@@ -24,6 +24,7 @@ services:
       - AGENT_SECRET=123123
       - RUN_MODE=master
       - LITEFS_IS_MASTER=true
+      - LITEFS_LISTEN_ADDR=:20202
     ports:
       - 8080:8080
       - 8888:8888
@@ -51,4 +52,5 @@ services:
       - MASTER_ENDPOINT=http://vorker-master:8888
       - NODE_NAME=agent-1
       - TUNNEL_HOST=vorker-master
-      - LITEFS_IS_MASTER=false
+      - LITEFS_IS_MASTER=false
+      - LITEFS_LISTEN_ADDR=:20202

docker-compose.yml

@@ -5,7 +5,7 @@ services:
     build: ./
     # image: vaalacat/vorker
     volumes:
-      - /tmp/workerd-master:/workerd # change to your own path
+      - /tmp/workerd-master:/var/lib/litefs # change to your own path
     environment:
       - WORKER_URL_SUFFIX=.example.com # worker url = SCHEME://WORKERNAME+WORKER_URL_SUFFIX eg: .example.com
       - SCHEME=http
@@ -14,6 +14,7 @@ services:
       - JWT_SECRET=123123
       - AGENT_SECRET=123123
       - RUN_MODE=master
+      - LITEFS_LISTEN_ADDR=:20202
     ports:
       - 8080:8080 # reverse proxy port
       - 8888:8888 # admin webui port
@@ -24,11 +25,12 @@ services:
     build: ./
     # image: vaalacat/vorker
     volumes:
-      - /tmp/workerd-agent:/workerd # change to your own path
+      - /tmp/workerd-agent:/var/lib/litefs # change to your own path
     environment:
       - WORKER_URL_SUFFIX=.example.com # worker url = SCHEME://WORKERNAME+WORKER_URL_SUFFIX eg: .example.com
       - AGENT_SECRET=123123 # must be same as master
       - RUN_MODE=agent
       - MASTER_ENDPOINT=http://vorker-master:8888
       - TUNNEL_RELAY_ENDPOINT=vorker-master:18080
       - NODE_NAME=agent-1 # must be different amone agents
+      - LITEFS_LISTEN_ADDR=:20201

litefs.yml

@@ -6,5 +6,7 @@ lease:
   type: "static"
   advertise-url: http://127.0.0.1:20202
   candidate: $LITEFS_IS_MASTER
+http:
+  addr: $LITEFS_LISTEN_ADDR
 # exec:
 #   - cmd: "/bin/sh"

models/init.go

@@ -15,8 +15,12 @@ func NodeWorkersInit() {
 	}
 	logrus.Infof("this node will init %d workers", len(workerRecords))
 	for _, worker := range workerRecords {
-		worker.Flush()
-		utils.GenWorkerConfig(worker.Worker)
+		if err := worker.Flush(); err != nil {
+			logrus.WithError(err).Errorf("init failed to flush worker, worker is: [%+v]", worker.Worker)
+		}
+		if err := utils.GenWorkerConfig(worker.Worker); err != nil {
+			logrus.WithError(err).Errorf("init failed to gen worker config, worker is: [%+v]", worker.Worker)
+		}
 		exec.ExecManager.RunCmd(worker.GetUID(), []string{})
 	}
 }

models/workerd.go

@@ -16,6 +16,7 @@ import (
 
 	"github.com/google/uuid"
 	"github.com/imroc/req/v3"
+	"github.com/samber/lo"
 	"github.com/sirupsen/logrus"
 	"google.golang.org/protobuf/proto"
 	"gorm.io/gorm"
@@ -224,7 +225,7 @@ func (w *Worker) Update() error {
 		w.Port = int32(port)
 		tunnel.GetClient().Delete(w.GetUID())
 		tunnel.GetClient().Add(w.GetUID(),
-			utils.WorkerHostPrefix(w.GetName()), int(w.GetPort()))
+			utils.WorkerHostPrefix(w.GetName()), port)
 		err = w.UpdateFile()
 		if err != nil {
 			return err
@@ -285,20 +286,15 @@ func (w *Worker) Flush() error {
 			defs.KeyWorkerProto: wp,
 		})
 	}
-	port, err := utils.GetAvailablePort(defs.DefaultHostName)
-	if err != nil {
-		return err
-	}
 	if len(w.TunnelID) == 0 {
 		w.TunnelID = uuid.New().String()
 	}
 
-	if err = w.DeleteFile(); err != nil {
+	if err := w.DeleteFile(); err != nil {
 		return err
 	}
 
-	w.Port = int32(port)
-	if err = w.Update(); err != nil {
+	if err := w.Update(); err != nil {
 		return err
 	}
 	return nil
@@ -350,9 +346,14 @@ func SyncWorkers(workerList *entities.WorkerList) error {
 		return err
 	}
 
+	oldWorkerUIDMap := lo.SliceToMap(oldWorkers, func(w *Worker) (string, bool) { return w.UID, true })
+
 	for _, worker := range workerList.Workers {
 		modelWorker := &Worker{Worker: worker}
 		UIDs = append(UIDs, worker.UID)
+		if _, ok := oldWorkerUIDMap[worker.UID]; ok {
+			continue
+		}
 
 		if err := modelWorker.Delete(); err != nil && err != gorm.ErrRecordNotFound {
 			logrus.WithError(err).Errorf("sync workers db delete error, worker is: %+v", worker)
@@ -377,6 +378,11 @@ func SyncWorkers(workerList *entities.WorkerList) error {
 			partialFail = true
 			continue
 		}
+		if err := utils.GenWorkerConfig(modelWorker.Worker); err != nil {
+			logrus.WithError(err).Errorf("sync workers gen config error, worker is: %+v", worker)
+			partialFail = true
+			continue
+		}
 	}
 
 	// delete workers that not in workerList

services/agent/sync.go

@@ -37,5 +37,5 @@ func SyncCall() error {
 	if err := models.SyncWorkers(workerList); err != nil {
 		return err
 	}
-	return workerd.GenCapnpConfig()
+	return nil
 }

services/workerd/capnp.go

@@ -5,6 +5,7 @@ import (
 	"path/filepath"
 	"vorker/conf"
 	"vorker/defs"
+	"vorker/entities"
 	"vorker/models"
 	"vorker/utils"
 
@@ -18,10 +19,17 @@ func GenCapnpConfig() error {
 	}
 
 	workerList := models.Trans2Entities(workerRecords)
-	fileMap := utils.BuildCapfile(workerList)
 
 	var hasError bool
 	for _, worker := range workerList {
+		w := &models.Worker{Worker: worker}
+		if err := w.Flush(); err != nil {
+			logrus.WithError(err).Errorf("failed to flush worker, worker is: %+v", worker)
+			hasError = true
+			continue
+		}
+		fileMap := utils.BuildCapfile([]*entities.Worker{w.Worker})
+
 		if fileContent, ok := fileMap[worker.GetUID()]; ok {
 			err := utils.WriteFile(
 				filepath.Join(

services/workerd/flush.go

@@ -57,6 +57,12 @@ func FlushAllEndpoint(c *gin.Context) {
 		return
 	}
 
+	if err := GenCapnpConfig(); err != nil {
+		common.RespErr(c, common.RespCodeInternalError, err.Error(), nil)
+		logrus.WithError(err).Error("gen capnp config error")
+		return
+	}
+
 	common.RespOK(c, "flush worker success", nil)
 }