mirror of
https://github.com/Rain-kl/OpenFlare.git
synced 2026-10-02 14:56:38 +08:00
feat: add OperationSetting component and integrate into settings page
- Introduced OperationSetting component for managing agent configurations. - Updated settings page to include a new tab for operation settings. - Implemented functionality to fetch and update agent parameters such as heartbeat interval, sync interval, and auto-update settings. docs: enhance deployment documentation for agent installation - Added detailed instructions for agent installation using a script. - Included examples for using discovery and agent tokens. - Updated sections on global discovery tokens and agent auto-update features. docs: revise design and development guidelines for V3 - Updated design document to reflect the current state and goals for V3. - Clarified development guidelines to focus on operational experience improvements. ci: add GitHub Actions workflow for agent releases - Created a new workflow to automate the release of agent binaries on GitHub. - Configured the workflow to build binaries for multiple platforms and publish them as releases. feat: implement self-update mechanism for agent - Added updater module to handle checking for and applying updates from GitHub releases. - Implemented logic to restart the agent after a successful update. chore: create install script for agent deployment - Developed a bash script to facilitate the installation of the ATSFlare agent. - The script supports automatic configuration and systemd service creation.
This commit is contained in:
@@ -14,6 +14,7 @@ import (
|
||||
"atsflare-agent/internal/nginx"
|
||||
"atsflare-agent/internal/state"
|
||||
syncservice "atsflare-agent/internal/sync"
|
||||
"atsflare-agent/internal/updater"
|
||||
)
|
||||
|
||||
func main() {
|
||||
@@ -58,6 +59,7 @@ func main() {
|
||||
NginxCertDir: cfg.NginxCertDir,
|
||||
}),
|
||||
}, stateStore),
|
||||
Updater: updater.New(),
|
||||
}
|
||||
|
||||
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
|
||||
|
||||
@@ -14,7 +14,7 @@ import (
|
||||
|
||||
type HeartbeatService interface {
|
||||
Register(ctx context.Context, payload protocol.NodePayload) (*protocol.RegisterNodeResponse, error)
|
||||
Heartbeat(ctx context.Context, payload protocol.NodePayload) error
|
||||
Heartbeat(ctx context.Context, payload protocol.NodePayload) (*protocol.AgentSettings, error)
|
||||
SetToken(token string)
|
||||
}
|
||||
|
||||
@@ -23,11 +23,19 @@ type SyncService interface {
|
||||
SyncOnce(ctx context.Context) error
|
||||
}
|
||||
|
||||
type Updater interface {
|
||||
CheckAndUpdate(ctx context.Context, repo string) error
|
||||
}
|
||||
|
||||
type Runner struct {
|
||||
Config *config.Config
|
||||
StateStore *state.Store
|
||||
HeartbeatService HeartbeatService
|
||||
SyncService SyncService
|
||||
Updater Updater
|
||||
|
||||
autoUpdate bool
|
||||
updateRepo string
|
||||
}
|
||||
|
||||
func (r *Runner) Run(ctx context.Context) error {
|
||||
@@ -43,10 +51,12 @@ func (r *Runner) Run(ctx context.Context) error {
|
||||
} else {
|
||||
log.Printf("agent startup sync completed")
|
||||
}
|
||||
if err = r.HeartbeatService.Heartbeat(ctx, r.nodePayload(nodeID)); err != nil {
|
||||
log.Printf("agent startup heartbeat failed: %v", err)
|
||||
settings, hbErr := r.HeartbeatService.Heartbeat(ctx, r.nodePayload(nodeID))
|
||||
if hbErr != nil {
|
||||
log.Printf("agent startup heartbeat failed: %v", hbErr)
|
||||
} else {
|
||||
log.Printf("agent startup heartbeat succeeded: node_id=%s", nodeID)
|
||||
r.applySettings(settings)
|
||||
}
|
||||
} else if err = r.tryRegister(ctx, &nodeID); err != nil {
|
||||
log.Printf("agent initial discovery register failed: %v", err)
|
||||
@@ -69,8 +79,15 @@ func (r *Runner) Run(ctx context.Context) error {
|
||||
}
|
||||
continue
|
||||
}
|
||||
if err = r.HeartbeatService.Heartbeat(ctx, r.nodePayload(nodeID)); err != nil {
|
||||
log.Printf("agent heartbeat failed: %v", err)
|
||||
settings, hbErr := r.HeartbeatService.Heartbeat(ctx, r.nodePayload(nodeID))
|
||||
if hbErr != nil {
|
||||
log.Printf("agent heartbeat failed: %v", hbErr)
|
||||
} else {
|
||||
if changed := r.applySettings(settings); changed {
|
||||
heartbeatTicker.Reset(r.Config.HeartbeatInterval.Duration())
|
||||
syncTicker.Reset(r.Config.SyncInterval.Duration())
|
||||
}
|
||||
r.tryAutoUpdate(ctx)
|
||||
}
|
||||
case <-syncTicker.C:
|
||||
if !r.hasAgentToken() {
|
||||
@@ -91,6 +108,41 @@ func (r *Runner) hasAgentToken() bool {
|
||||
return strings.TrimSpace(r.Config.AgentToken) != ""
|
||||
}
|
||||
|
||||
func (r *Runner) applySettings(settings *protocol.AgentSettings) bool {
|
||||
if settings == nil {
|
||||
return false
|
||||
}
|
||||
changed := false
|
||||
if settings.HeartbeatInterval > 0 {
|
||||
newInterval := config.MillisecondDuration(time.Duration(settings.HeartbeatInterval) * time.Millisecond)
|
||||
if newInterval != r.Config.HeartbeatInterval {
|
||||
log.Printf("agent heartbeat interval updated: %s -> %s", r.Config.HeartbeatInterval, newInterval)
|
||||
r.Config.HeartbeatInterval = newInterval
|
||||
changed = true
|
||||
}
|
||||
}
|
||||
if settings.SyncInterval > 0 {
|
||||
newInterval := config.MillisecondDuration(time.Duration(settings.SyncInterval) * time.Millisecond)
|
||||
if newInterval != r.Config.SyncInterval {
|
||||
log.Printf("agent sync interval updated: %s -> %s", r.Config.SyncInterval, newInterval)
|
||||
r.Config.SyncInterval = newInterval
|
||||
changed = true
|
||||
}
|
||||
}
|
||||
r.autoUpdate = settings.AutoUpdate
|
||||
r.updateRepo = settings.UpdateRepo
|
||||
return changed
|
||||
}
|
||||
|
||||
func (r *Runner) tryAutoUpdate(ctx context.Context) {
|
||||
if !r.autoUpdate || r.Updater == nil || r.updateRepo == "" {
|
||||
return
|
||||
}
|
||||
if err := r.Updater.CheckAndUpdate(ctx, r.updateRepo); err != nil {
|
||||
log.Printf("agent auto-update check failed: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
func (r *Runner) tryRegister(ctx context.Context, nodeID *string) error {
|
||||
if strings.TrimSpace(r.Config.DiscoveryToken) == "" {
|
||||
return errors.New("agent_token 为空且未配置 discovery_token")
|
||||
|
||||
@@ -32,7 +32,7 @@ func (f *fakeHeartbeatService) Register(ctx context.Context, payload protocol.No
|
||||
return f.registerResp, f.registerErr
|
||||
}
|
||||
|
||||
func (f *fakeHeartbeatService) Heartbeat(ctx context.Context, payload protocol.NodePayload) error {
|
||||
func (f *fakeHeartbeatService) Heartbeat(ctx context.Context, payload protocol.NodePayload) (*protocol.AgentSettings, error) {
|
||||
f.mu.Lock()
|
||||
f.heartbeatCalls++
|
||||
callIndex := f.heartbeatCalls
|
||||
@@ -45,7 +45,7 @@ func (f *fakeHeartbeatService) Heartbeat(ctx context.Context, payload protocol.N
|
||||
if onHeartbeat != nil {
|
||||
onHeartbeat(callIndex)
|
||||
}
|
||||
return err
|
||||
return nil, err
|
||||
}
|
||||
|
||||
func (f *fakeHeartbeatService) SetToken(token string) {
|
||||
|
||||
@@ -1,3 +1,3 @@
|
||||
package config
|
||||
|
||||
const AgentVersion = "0.2.0"
|
||||
const AgentVersion = "0.3.0"
|
||||
|
||||
@@ -8,7 +8,7 @@ import (
|
||||
|
||||
type Client interface {
|
||||
RegisterNode(ctx context.Context, payload protocol.NodePayload) (*protocol.RegisterNodeResponse, error)
|
||||
Heartbeat(ctx context.Context, payload protocol.NodePayload) error
|
||||
Heartbeat(ctx context.Context, payload protocol.NodePayload) (*protocol.AgentSettings, error)
|
||||
SetToken(token string)
|
||||
}
|
||||
|
||||
@@ -24,7 +24,7 @@ func (s *Service) Register(ctx context.Context, payload protocol.NodePayload) (*
|
||||
return s.client.RegisterNode(ctx, payload)
|
||||
}
|
||||
|
||||
func (s *Service) Heartbeat(ctx context.Context, payload protocol.NodePayload) error {
|
||||
func (s *Service) Heartbeat(ctx context.Context, payload protocol.NodePayload) (*protocol.AgentSettings, error) {
|
||||
return s.client.Heartbeat(ctx, payload)
|
||||
}
|
||||
|
||||
|
||||
@@ -42,8 +42,15 @@ func (c *Client) RegisterNode(ctx context.Context, payload protocol.NodePayload)
|
||||
return &resp.Data, nil
|
||||
}
|
||||
|
||||
func (c *Client) Heartbeat(ctx context.Context, payload protocol.NodePayload) error {
|
||||
return c.postJSON(ctx, "/api/agent/nodes/heartbeat", payload, nil)
|
||||
func (c *Client) Heartbeat(ctx context.Context, payload protocol.NodePayload) (*protocol.AgentSettings, error) {
|
||||
resp := protocol.HeartbeatAPIResponse{}
|
||||
if err := c.postJSON(ctx, "/api/agent/nodes/heartbeat", payload, &resp); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if !resp.Success {
|
||||
return nil, errors.New(resp.Message)
|
||||
}
|
||||
return resp.AgentSettings, nil
|
||||
}
|
||||
|
||||
func (c *Client) GetActiveConfig(ctx context.Context) (*protocol.ActiveConfigResponse, error) {
|
||||
|
||||
@@ -6,6 +6,20 @@ type APIResponse[T any] struct {
|
||||
Data T `json:"data"`
|
||||
}
|
||||
|
||||
type HeartbeatAPIResponse struct {
|
||||
Success bool `json:"success"`
|
||||
Message string `json:"message"`
|
||||
Data any `json:"data"`
|
||||
AgentSettings *AgentSettings `json:"agent_settings,omitempty"`
|
||||
}
|
||||
|
||||
type AgentSettings struct {
|
||||
HeartbeatInterval int `json:"heartbeat_interval"`
|
||||
SyncInterval int `json:"sync_interval"`
|
||||
AutoUpdate bool `json:"auto_update"`
|
||||
UpdateRepo string `json:"update_repo"`
|
||||
}
|
||||
|
||||
type NodePayload struct {
|
||||
NodeID string `json:"node_id"`
|
||||
Name string `json:"name"`
|
||||
|
||||
@@ -0,0 +1,195 @@
|
||||
package updater
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"log"
|
||||
"net/http"
|
||||
"os"
|
||||
"runtime"
|
||||
"strings"
|
||||
"syscall"
|
||||
"time"
|
||||
|
||||
"atsflare-agent/internal/config"
|
||||
)
|
||||
|
||||
type Service struct {
|
||||
httpClient *http.Client
|
||||
lastCheckTag string
|
||||
}
|
||||
|
||||
func New() *Service {
|
||||
return &Service{
|
||||
httpClient: &http.Client{Timeout: 30 * time.Second},
|
||||
}
|
||||
}
|
||||
|
||||
type githubRelease struct {
|
||||
TagName string `json:"tag_name"`
|
||||
Assets []githubAsset `json:"assets"`
|
||||
}
|
||||
|
||||
type githubAsset struct {
|
||||
Name string `json:"name"`
|
||||
BrowserDownloadURL string `json:"browser_download_url"`
|
||||
}
|
||||
|
||||
func (s *Service) CheckAndUpdate(ctx context.Context, repo string) error {
|
||||
release, err := s.getLatestRelease(ctx, repo)
|
||||
if err != nil {
|
||||
return fmt.Errorf("check latest release: %w", err)
|
||||
}
|
||||
if release == nil || release.TagName == "" {
|
||||
return nil
|
||||
}
|
||||
|
||||
remoteVersion := normalizeVersion(release.TagName)
|
||||
localVersion := normalizeVersion(config.AgentVersion)
|
||||
|
||||
if remoteVersion == localVersion || remoteVersion == s.lastCheckTag {
|
||||
return nil
|
||||
}
|
||||
if !isNewer(localVersion, remoteVersion) {
|
||||
s.lastCheckTag = remoteVersion
|
||||
return nil
|
||||
}
|
||||
|
||||
log.Printf("agent update available: %s -> %s", localVersion, remoteVersion)
|
||||
assetName := fmt.Sprintf("atsflare-agent-%s-%s", runtime.GOOS, runtime.GOARCH)
|
||||
|
||||
var downloadURL string
|
||||
for _, asset := range release.Assets {
|
||||
if asset.Name == assetName {
|
||||
downloadURL = asset.BrowserDownloadURL
|
||||
break
|
||||
}
|
||||
}
|
||||
if downloadURL == "" {
|
||||
s.lastCheckTag = remoteVersion
|
||||
return fmt.Errorf("no matching asset %q in release %s", assetName, release.TagName)
|
||||
}
|
||||
|
||||
execPath, err := os.Executable()
|
||||
if err != nil {
|
||||
return fmt.Errorf("get executable path: %w", err)
|
||||
}
|
||||
if err = s.downloadAndReplace(ctx, downloadURL, execPath); err != nil {
|
||||
return fmt.Errorf("download and replace: %w", err)
|
||||
}
|
||||
|
||||
log.Printf("agent binary updated, restarting...")
|
||||
return s.restart(execPath)
|
||||
}
|
||||
|
||||
func (s *Service) getLatestRelease(ctx context.Context, repo string) (*githubRelease, error) {
|
||||
url := fmt.Sprintf("https://api.github.com/repos/%s/releases/latest", repo)
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
req.Header.Set("Accept", "application/vnd.github+json")
|
||||
|
||||
resp, err := s.httpClient.Do(req)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
if resp.StatusCode == http.StatusNotFound {
|
||||
return nil, nil
|
||||
}
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return nil, fmt.Errorf("github api returned %s", resp.Status)
|
||||
}
|
||||
|
||||
var release githubRelease
|
||||
if err = json.NewDecoder(resp.Body).Decode(&release); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return &release, nil
|
||||
}
|
||||
|
||||
func (s *Service) downloadAndReplace(ctx context.Context, url string, targetPath string) error {
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
resp, err := s.httpClient.Do(req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return fmt.Errorf("download returned %s", resp.Status)
|
||||
}
|
||||
|
||||
tmpPath := targetPath + ".update"
|
||||
tmpFile, err := os.OpenFile(tmpPath, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, 0o755)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err = io.Copy(tmpFile, resp.Body); err != nil {
|
||||
tmpFile.Close()
|
||||
os.Remove(tmpPath)
|
||||
return err
|
||||
}
|
||||
tmpFile.Close()
|
||||
|
||||
backupPath := targetPath + ".bak"
|
||||
os.Remove(backupPath)
|
||||
if err = os.Rename(targetPath, backupPath); err != nil {
|
||||
os.Remove(tmpPath)
|
||||
return fmt.Errorf("backup current binary: %w", err)
|
||||
}
|
||||
if err = os.Rename(tmpPath, targetPath); err != nil {
|
||||
// Attempt to restore backup
|
||||
os.Rename(backupPath, targetPath)
|
||||
return fmt.Errorf("replace binary: %w", err)
|
||||
}
|
||||
os.Remove(backupPath)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (s *Service) restart(execPath string) error {
|
||||
argv := os.Args
|
||||
if err := syscall.Exec(execPath, argv, os.Environ()); err != nil {
|
||||
return fmt.Errorf("exec restart: %w", err)
|
||||
}
|
||||
return errors.New("unreachable after exec")
|
||||
}
|
||||
|
||||
func normalizeVersion(v string) string {
|
||||
v = strings.TrimSpace(v)
|
||||
v = strings.TrimPrefix(v, "v")
|
||||
return v
|
||||
}
|
||||
|
||||
func isNewer(local, remote string) bool {
|
||||
localParts := strings.Split(local, ".")
|
||||
remoteParts := strings.Split(remote, ".")
|
||||
maxLen := len(localParts)
|
||||
if len(remoteParts) > maxLen {
|
||||
maxLen = len(remoteParts)
|
||||
}
|
||||
for i := 0; i < maxLen; i++ {
|
||||
lp, rp := "0", "0"
|
||||
if i < len(localParts) {
|
||||
lp = localParts[i]
|
||||
}
|
||||
if i < len(remoteParts) {
|
||||
rp = remoteParts[i]
|
||||
}
|
||||
if rp > lp {
|
||||
return true
|
||||
}
|
||||
if rp < lp {
|
||||
return false
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
Reference in New Issue
Block a user