集群可以设置systemd系统服务

internal/nodes/api_stream.go

@@ -7,13 +7,16 @@ import (
 	"github.com/TeaOSLab/EdgeCommon/pkg/rpc/pb"
 	"github.com/TeaOSLab/EdgeCommon/pkg/serverconfigs"
 	"github.com/TeaOSLab/EdgeNode/internal/caches"
+	teaconst "github.com/TeaOSLab/EdgeNode/internal/const"
 	"github.com/TeaOSLab/EdgeNode/internal/errors"
 	"github.com/TeaOSLab/EdgeNode/internal/events"
 	"github.com/TeaOSLab/EdgeNode/internal/iplibrary"
 	"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
 	"github.com/TeaOSLab/EdgeNode/internal/rpc"
+	"github.com/TeaOSLab/EdgeNode/internal/utils"
 	"io"
 	"net/http"
+	"os/exec"
 	"strconv"
 	"strings"
 	"sync"
@@ -98,6 +101,8 @@ func (this *APIStream) loop() error {
 			err = this.handleConfigChanged(message)
 		case messageconfigs.MessageCodeIPListChanged: // IPList变化
 			err = this.handleIPListChanged(message)
+		case messageconfigs.MessageCodeCheckSystemdService: // 检查Systemd服务
+			err = this.handleCheckSystemdService(message)
 		default:
 			err = this.handleUnknownMessage(message)
 		}
@@ -373,6 +378,7 @@ func (this *APIStream) handlePreheatCache(message *pb.NodeStreamMessage) error {
 			}()
 
 			// 检查最大内容长度
+			// TODO 需要解决Chunked Transfer Encoding的长度判断问题
 			maxSize := storage.Policy().MaxSizeBytes()
 			if maxSize > 0 && resp.ContentLength > maxSize {
 				locker.Lock()
@@ -461,6 +467,34 @@ func (this *APIStream) handleIPListChanged(message *pb.NodeStreamMessage) error
 	return nil
 }
 
+// 检查Systemd服务
+func (this *APIStream) handleCheckSystemdService(message *pb.NodeStreamMessage) error {
+	systemctl, err := exec.LookPath("systemctl")
+	if err != nil {
+		this.replyFail(message.RequestId, "'systemctl' not found")
+		return nil
+	}
+	if len(systemctl) == 0 {
+		this.replyFail(message.RequestId, "'systemctl' not found")
+		return nil
+	}
+
+	cmd := utils.NewCommandExecutor()
+	shortName := teaconst.SystemdServiceName
+	cmd.Add(systemctl, "is-enabled", shortName)
+	output, err := cmd.Run()
+	if err != nil {
+		this.replyFail(message.RequestId, "'systemctl' command error: " + err.Error())
+		return nil
+	}
+	if output == "enabled" {
+		this.replyOk(message.RequestId, "ok")
+	} else {
+		this.replyFail(message.RequestId, "not installed")
+	}
+	return nil
+}
+
 // 处理未知消息
 func (this *APIStream) handleUnknownMessage(message *pb.NodeStreamMessage) error {
 	this.replyFail(message.RequestId, "unknown message code '"+message.Code+"'")

internal/nodes/node.go

@@ -16,7 +16,7 @@ import (
 	"github.com/TeaOSLab/EdgeNode/internal/utils"
 	"github.com/go-yaml/yaml"
 	"github.com/iwind/TeaGo/Tea"
-	tealogs "github.com/iwind/TeaGo/logs"
+	"github.com/iwind/TeaGo/logs"
 	"io/ioutil"
 	"net"
 	"os"
@@ -69,10 +69,24 @@ func (this *Node) Start() {
 	}
 
 	// 读取API配置
-	err = this.syncConfig(false)
-	if err != nil {
-		remotelogs.Error("NODE", err.Error())
-		return
+	tryTimes := 0
+	for {
+		err = this.syncConfig(false)
+		if err != nil {
+			tryTimes++
+
+			if tryTimes%10 == 0 {
+				remotelogs.Error("NODE", err.Error())
+			}
+			time.Sleep(1 * time.Second)
+
+			// 不做长时间的无意义的重试
+			if tryTimes > 1000 {
+				return
+			}
+		} else {
+			break
+		}
 	}
 
 	// 启动同步计时器
@@ -292,12 +306,12 @@ func (this *Node) checkClusterConfig() error {
 		return err
 	}
 
-	tealogs.Println("[NODE]registering node ...")
+	logs.Println("[NODE]registering node ...")
 	resp, err := rpcClient.NodeRPC().RegisterClusterNode(rpcClient.ClusterContext(config.ClusterId, config.Secret), &pb.RegisterClusterNodeRequest{Name: HOSTNAME})
 	if err != nil {
 		return err
 	}
-	tealogs.Println("[NODE]registered successfully")
+	logs.Println("[NODE]registered successfully")
 
 	// 写入到配置文件中
 	if len(resp.Endpoints) == 0 {
@@ -312,12 +326,12 @@ func (this *Node) checkClusterConfig() error {
 		NodeId: resp.UniqueId,
 		Secret: resp.Secret,
 	}
-	tealogs.Println("[NODE]writing 'configs/api.yaml' ...")
+	logs.Println("[NODE]writing 'configs/api.yaml' ...")
 	err = apiConfig.WriteFile(Tea.ConfigFile("api.yaml"))
 	if err != nil {
 		return err
 	}
-	tealogs.Println("[NODE]wrote 'configs/api.yaml' successfully")
+	logs.Println("[NODE]wrote 'configs/api.yaml' successfully")
 
 	return nil
 }

internal/nodes/system_services.go

@@ -0,0 +1,118 @@
+package nodes
+
+import (
+	"bytes"
+	"encoding/json"
+	"errors"
+	"github.com/TeaOSLab/EdgeCommon/pkg/nodeconfigs"
+	teaconst "github.com/TeaOSLab/EdgeNode/internal/const"
+	"github.com/TeaOSLab/EdgeNode/internal/events"
+	"github.com/TeaOSLab/EdgeNode/internal/remotelogs"
+	"github.com/TeaOSLab/EdgeNode/internal/utils"
+	"github.com/iwind/TeaGo/maps"
+	"io/ioutil"
+	"os"
+	"os/exec"
+	"runtime"
+)
+
+func init() {
+	var manager = NewSystemServiceManager()
+	events.On(events.EventReload, func() {
+		err := manager.Setup()
+		if err != nil {
+			remotelogs.Error("SYSTEM_SERVICE", "setup system services failed: "+err.Error())
+		}
+	})
+}
+
+// 系统服务管理
+type SystemServiceManager struct {
+}
+
+func NewSystemServiceManager() *SystemServiceManager {
+	return &SystemServiceManager{}
+}
+
+func (this *SystemServiceManager) Setup() error {
+	if sharedNodeConfig == nil || !sharedNodeConfig.IsOn {
+		return nil
+	}
+
+	if len(sharedNodeConfig.SystemServices) == 0 {
+		return nil
+	}
+
+	systemdParams, ok := sharedNodeConfig.SystemServices[nodeconfigs.SystemServiceTypeSystemd]
+	if ok {
+		err := this.setupSystemd(systemdParams)
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func (this *SystemServiceManager) setupSystemd(params maps.Map) error {
+	// 只有在Linux下运行
+	if runtime.GOOS != "linux" {
+		return nil
+	}
+
+	if params == nil {
+		params = maps.Map{}
+	}
+	data, err := json.Marshal(params)
+	if err != nil {
+		return err
+	}
+	config := &nodeconfigs.SystemdServiceConfig{}
+	err = json.Unmarshal(data, config)
+	if err != nil {
+		return err
+	}
+
+	// 检查当前的service
+	systemctl, err := exec.LookPath("systemctl")
+	if err != nil {
+		return err
+	}
+	if len(systemctl) == 0 {
+		return errors.New("can not find 'systemctl' on the system")
+	}
+	cmd := utils.NewCommandExecutor()
+	shortName := teaconst.SystemdServiceName
+	cmd.Add(systemctl, "is-enabled", shortName)
+	output, err := cmd.Run()
+	if err != nil {
+		return err
+	}
+	if config.IsOn {
+		exe, err := os.Executable()
+		if err != nil {
+			return err
+		}
+
+		if output == "enabled" {
+			// 检查文件路径是否变化
+			data, err := ioutil.ReadFile("/etc/systemd/system/" + teaconst.SystemdServiceName + ".service")
+			if err == nil && bytes.Index(data, []byte(exe)) > 0 {
+				return nil
+			}
+		}
+		manager := utils.NewServiceManager(shortName, teaconst.ProductName)
+		err = manager.Install(exe, []string{})
+		if err != nil {
+			return err
+		}
+	} else {
+		manager := utils.NewServiceManager(shortName, teaconst.ProductName)
+		err = manager.Uninstall()
+		if err != nil {
+			return err
+		}
+	}
+
+	return nil
+}